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
59
47
if sys.version_info.major == 2:
62
50
locale.setlocale(locale.LC_ALL, "")
56
"LastCheckedOK": "Last Successful Check",
57
"LastApprovalRequest": "Last Approval Request",
59
"Interval": "Interval",
61
"Fingerprint": "Fingerprint",
63
"CheckerRunning": "Check Is Running",
64
"LastEnabled": "Last Enabled",
65
"ApprovalPending": "Approval Is Pending",
66
"ApprovedByDefault": "Approved By Default",
67
"ApprovalDelay": "Approval Delay",
68
"ApprovalDuration": "Approval Duration",
70
"ExtendedTimeout": "Extended Timeout",
72
"LastCheckerStatus": "Last Checker Status",
64
74
defaultkeywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
65
75
domain = "se.recompile"
66
76
busname = domain + ".Mandos"
104
112
datetime.timedelta(0, 330)
105
113
>>> rfc3339_duration_to_delta("P1DT3M20S")
106
114
datetime.timedelta(1, 200)
107
>>> # Can not be empty:
108
>>> rfc3339_duration_to_delta("")
109
Traceback (most recent call last):
111
ValueError: Invalid RFC 3339 duration: u''
112
>>> # Must start with "P":
113
>>> rfc3339_duration_to_delta("1D")
114
Traceback (most recent call last):
116
ValueError: Invalid RFC 3339 duration: u'1D'
117
>>> # Must use correct order
118
>>> rfc3339_duration_to_delta("PT1S2M")
119
Traceback (most recent call last):
121
ValueError: Invalid RFC 3339 duration: u'PT1S2M'
122
>>> # Time needs time marker
123
>>> rfc3339_duration_to_delta("P1H2S")
124
Traceback (most recent call last):
126
ValueError: Invalid RFC 3339 duration: u'P1H2S'
127
>>> # Weeks can not be combined with anything else
128
>>> rfc3339_duration_to_delta("P1D2W")
129
Traceback (most recent call last):
131
ValueError: Invalid RFC 3339 duration: u'P1D2W'
132
>>> rfc3339_duration_to_delta("P2W2H")
133
Traceback (most recent call last):
135
ValueError: Invalid RFC 3339 duration: u'P2W2H'
138
117
# Parsing an RFC 3339 duration with regular expressions is not
218
197
def string_to_delta(interval):
219
198
"""Parse a string and return a datetime.timedelta
200
>>> string_to_delta('7d')
201
datetime.timedelta(7)
202
>>> string_to_delta('60s')
203
datetime.timedelta(0, 60)
204
>>> string_to_delta('60m')
205
datetime.timedelta(0, 3600)
206
>>> string_to_delta('24h')
207
datetime.timedelta(1)
208
>>> string_to_delta('1w')
209
datetime.timedelta(7)
210
>>> string_to_delta('5m 30s')
211
datetime.timedelta(0, 330)
223
215
return rfc3339_duration_to_delta(interval)
224
except ValueError as e:
225
log.warning("%s - Parsing as pre-1.6.1 interval instead",
227
return parse_pre_1_6_1_interval(interval)
230
def parse_pre_1_6_1_interval(interval):
231
"""Parse an interval string as documented by Mandos before 1.6.1, and
232
return a datetime.timedelta
233
>>> parse_pre_1_6_1_interval('7d')
234
datetime.timedelta(7)
235
>>> parse_pre_1_6_1_interval('60s')
236
datetime.timedelta(0, 60)
237
>>> parse_pre_1_6_1_interval('60m')
238
datetime.timedelta(0, 3600)
239
>>> parse_pre_1_6_1_interval('24h')
240
datetime.timedelta(1)
241
>>> parse_pre_1_6_1_interval('1w')
242
datetime.timedelta(7)
243
>>> parse_pre_1_6_1_interval('5m 30s')
244
datetime.timedelta(0, 330)
245
>>> parse_pre_1_6_1_interval('')
246
datetime.timedelta(0)
247
>>> # Ignore unknown characters, allow any order and repetitions
248
>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')
249
datetime.timedelta(2, 480, 18000)
253
219
value = datetime.timedelta(0)
254
220
regexp = re.compile(r"(\d+)([dsmhw]?)")
272
238
def print_clients(clients, keywords):
273
print('\n'.join(TableOfClients(clients, keywords).rows()))
275
class TableOfClients(object):
278
"Enabled": "Enabled",
279
"Timeout": "Timeout",
280
"LastCheckedOK": "Last Successful Check",
281
"LastApprovalRequest": "Last Approval Request",
282
"Created": "Created",
283
"Interval": "Interval",
285
"Fingerprint": "Fingerprint",
287
"CheckerRunning": "Check Is Running",
288
"LastEnabled": "Last Enabled",
289
"ApprovalPending": "Approval Is Pending",
290
"ApprovedByDefault": "Approved By Default",
291
"ApprovalDelay": "Approval Delay",
292
"ApprovalDuration": "Approval Duration",
293
"Checker": "Checker",
294
"ExtendedTimeout": "Extended Timeout",
295
"Expires": "Expires",
296
"LastCheckerStatus": "Last Checker Status",
299
def __init__(self, clients, keywords, tablewords=None):
300
self.clients = clients
301
self.keywords = keywords
302
if tablewords is not None:
303
self.tablewords = tablewords
306
format_string = self.row_formatting_string()
307
rows = [self.header_line(format_string)]
308
rows.extend(self.client_line(client, format_string)
309
for client in self.clients)
312
def row_formatting_string(self):
313
"Format string used to format table rows"
314
return " ".join("{{{key}:{width}}}".format(
315
width=max(len(self.tablewords[key]),
316
max(len(self.string_from_client(client, key))
317
for client in self.clients)),
319
for key in self.keywords)
321
def string_from_client(self, client, key):
322
return self.valuetostring(client[key], key)
325
239
def valuetostring(value, keyword):
326
if isinstance(value, dbus.Boolean):
240
if type(value) is dbus.Boolean:
327
241
return "Yes" if value else "No"
328
242
if keyword in ("Timeout", "Interval", "ApprovalDelay",
329
243
"ApprovalDuration", "ExtendedTimeout"):
330
244
return milliseconds_to_string(value)
331
245
return str(value)
333
def header_line(self, format_string):
334
return format_string.format(**self.tablewords)
336
def client_line(self, client, format_string):
337
return format_string.format(
338
**{key: self.string_from_client(client, key)
339
for key in self.keywords})
247
# Create format string to print table rows
248
format_string = " ".join("{{{key}:{width}}}".format(
249
width=max(len(tablewords[key]),
250
max(len(valuetostring(client[key], key))
251
for client in clients)),
255
print(format_string.format(**tablewords))
256
for client in clients:
258
.format(**{key: valuetostring(client[key], key)
259
for key in keywords}))
342
262
def has_actions(options):
568
490
dbus_interface=client_interface)
571
class Test_milliseconds_to_string(unittest.TestCase):
573
self.assertEqual(milliseconds_to_string(93785000),
575
def test_no_days(self):
576
self.assertEqual(milliseconds_to_string(7385000), "02:03:05")
577
def test_all_zero(self):
578
self.assertEqual(milliseconds_to_string(0), "00:00:00")
579
def test_no_fractional_seconds(self):
580
self.assertEqual(milliseconds_to_string(400), "00:00:00")
581
self.assertEqual(milliseconds_to_string(900), "00:00:00")
582
self.assertEqual(milliseconds_to_string(1900), "00:00:01")
584
class Test_string_to_delta(unittest.TestCase):
585
def test_handles_basic_rfc3339(self):
586
self.assertEqual(string_to_delta("PT2H"),
587
datetime.timedelta(0, 7200))
588
def test_falls_back_to_pre_1_6_1_with_warning(self):
589
# assertLogs only exists in Python 3.4
590
if hasattr(self, "assertLogs"):
591
with self.assertLogs(log, logging.WARNING):
592
value = string_to_delta("2h")
594
value = string_to_delta("2h")
595
self.assertEqual(value, datetime.timedelta(0, 7200))
597
class Test_TableOfClients(unittest.TestCase):
603
"Bool": "A D-BUS Boolean",
604
"NonDbusBoolean": "A Non-D-BUS Boolean",
605
"Integer": "An Integer",
606
"Timeout": "Timedelta 1",
607
"Interval": "Timedelta 2",
608
"ApprovalDelay": "Timedelta 3",
609
"ApprovalDuration": "Timedelta 4",
610
"ExtendedTimeout": "Timedelta 5",
611
"String": "A String",
613
self.keywords = ["Attr1", "AttrTwo"]
619
"Bool": dbus.Boolean(False),
620
"NonDbusBoolean": False,
624
"ApprovalDelay": 2000,
625
"ApprovalDuration": 3000,
626
"ExtendedTimeout": 4000,
633
"Bool": dbus.Boolean(True),
634
"NonDbusBoolean": True,
637
"Interval": 93786000,
638
"ApprovalDelay": 93787000,
639
"ApprovalDuration": 93788000,
640
"ExtendedTimeout": 93789000,
641
"String": "A huge string which will not fit," * 10,
644
def test_short_header(self):
645
rows = TableOfClients(self.clients, self.keywords,
646
self.tablewords).rows()
651
self.assertEqual(rows, expected_rows)
652
def test_booleans(self):
653
keywords = ["Bool", "NonDbusBoolean"]
654
rows = TableOfClients(self.clients, keywords,
655
self.tablewords).rows()
657
"A D-BUS Boolean A Non-D-BUS Boolean",
661
self.assertEqual(rows, expected_rows)
662
def test_milliseconds_detection(self):
663
keywords = ["Integer", "Timeout", "Interval", "ApprovalDelay",
664
"ApprovalDuration", "ExtendedTimeout"]
665
rows = TableOfClients(self.clients, keywords,
666
self.tablewords).rows()
668
An Integer Timedelta 1 Timedelta 2 Timedelta 3 Timedelta 4 Timedelta 5
669
0 00:00:00 00:00:01 00:00:02 00:00:03 00:00:04
670
1 1T02:03:05 1T02:03:06 1T02:03:07 1T02:03:08 1T02:03:09
673
self.assertEqual(rows, expected_rows)
674
def test_empty_and_long_string_values(self):
675
keywords = ["String"]
676
rows = TableOfClients(self.clients, keywords,
677
self.tablewords).rows()
681
A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,
684
self.assertEqual(rows, expected_rows)
688
493
def should_only_run_tests():
689
494
parser = argparse.ArgumentParser(add_help=False)
690
495
parser.add_argument("--check", action='store_true')