/mandos/release : revision 237.7.592

44

import logging

45

import io

46

import tempfile

47

import contextlib

47

48

49

import dbus

49

50

299

300

"""Abstract class for Actions for setting one client property"""

300

301

def run_on_one_client(self, client, properties):

301

302

"""Set the Client's D-Bus property"""

303

log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,

304

client.__dbus_object_path__,

305

dbus.PROPERTIES_IFACE, client_interface,

306

self.property, self.value_to_set

307

if not isinstance(self.value_to_set, dbus.Boolean)

308

else bool(self.value_to_set))

302

309

client.Set(client_interface, self.property, self.value_to_set,

303

310

dbus_interface=dbus.PROPERTIES_IFACE)

304

311

316

323

@value_to_set.setter

317

324

def value_to_set(self, value):

318

325

"""When setting, convert value to a datetime.timedelta"""

319

self._vts = string_to_delta(value).total_seconds() * 1000

326

self._vts = int(round(value.total_seconds() * 1000))

320

327

321

328

# Actual (non-abstract) command classes

322

329

430

437

431

438

class RemoveCmd(Command):

432

439

def run_on_one_client(self, client, properties):

440

log.debug("D-Bus: %s:%s:%s.RemoveClient(%r)", busname,

441

server_path, server_interface,

442

str(client.__dbus_object_path__))

433

443

self.mandos.RemoveClient(client.__dbus_object_path__)

434

444

435

445

class ApproveCmd(Command):

436

446

def run_on_one_client(self, client, properties):

447

log.debug("D-Bus: %s:%s.Approve(True)",

448

client.__dbus_object_path__, client_interface)

437

449

client.Approve(dbus.Boolean(True),

438

450

dbus_interface=client_interface)

439

451

440

452

class DenyCmd(Command):

441

453

def run_on_one_client(self, client, properties):

454

log.debug("D-Bus: %s:%s.Approve(False)",

455

client.__dbus_object_path__, client_interface)

442

456

client.Approve(dbus.Boolean(False),

443

457

dbus_interface=client_interface)

444

458

505

519

MillisecondsValueArgumentMixIn):

506

520

property = "ApprovalDuration"

507

521

508

def has_actions(options):

509

return any((options.enable,

510

options.disable,

511

options.bump_timeout,

512

options.start_checker,

513

options.stop_checker,

514

options.is_enabled,

515

options.remove,

516

options.checker is not None,

517

options.timeout is not None,

518

options.extended_timeout is not None,

519

options.interval is not None,

520

options.approved_by_default is not None,

521

options.approval_delay is not None,

522

options.approval_duration is not None,

523

options.host is not None,

524

options.secret is not None,

525

options.approve,

526

options.deny))

527

528

522

def add_command_line_options(parser):

529

523

parser.add_argument("--version", action="version",

530

524

version="%(prog)s {}".format(version),

556

550

help="Remove client")

557

551

parser.add_argument("-c", "--checker",

558

552

help="Set checker command for client")

559

parser.add_argument("-t", "--timeout",

553

parser.add_argument("-t", "--timeout", type=string_to_delta,

560

554

help="Set timeout for client")

561

parser.add_argument("--extended-timeout",

555

parser.add_argument("--extended-timeout", type=string_to_delta,

562

556

help="Set extended timeout for client")

563

parser.add_argument("-i", "--interval",

557

parser.add_argument("-i", "--interval", type=string_to_delta,

564

558

help="Set checker interval for client")

565

559

approve_deny_default = parser.add_mutually_exclusive_group()

566

560

approve_deny_default.add_argument(

571

565

"--deny-by-default", action="store_false",

572

566

dest="approved_by_default",

573

567

help="Set client to be denied by default")

574

parser.add_argument("--approval-delay",

568

parser.add_argument("--approval-delay", type=string_to_delta,

575

569

help="Set delay before client approve/deny")

576

parser.add_argument("--approval-duration",

570

parser.add_argument("--approval-duration", type=string_to_delta,

577

571

help="Set duration of one client approval")

578

572

parser.add_argument("-H", "--host", help="Set host for client")

579

573

parser.add_argument("-s", "--secret",

585

579

help="Approve any current client request")

586

580

approve_deny.add_argument("-D", "--deny", action="store_true",

587

581

help="Deny any current client request")

582

parser.add_argument("--debug", action="store_true",

583

help="Debug mode (show D-Bus commands)")

588

584

parser.add_argument("--check", action="store_true",

589

585

help="Run self-test")

590

586

parser.add_argument("client", nargs="*", help="Client name")

615

611

if options.is_enabled:

616

612

commands.append(IsEnabledCmd())

617

613

618

if options.remove:

619

commands.append(RemoveCmd())

620

621

614

if options.checker is not None:

622

615

commands.append(SetCheckerCmd(options.checker))

623

616

656

649

if options.deny:

657

650

commands.append(DenyCmd())

658

651

652

if options.remove:

653

commands.append(RemoveCmd())

654

659

655

# If no command option has been given, show table of clients,

660

656

# optionally verbosely

661

657

if not commands:

664

660

return commands

665

661

666

662

667

def main():

668

parser = argparse.ArgumentParser()

669

670

add_command_line_options(parser)

671

672

options = parser.parse_args()

663

def check_option_syntax(parser, options):

664

"""Apply additional restrictions on options, not expressible in

665

argparse"""

666

667

def has_actions(options):

668

return any((options.enable,

669

options.disable,

670

options.bump_timeout,

671

options.start_checker,

672

options.stop_checker,

673

options.is_enabled,

674

options.remove,

675

options.checker is not None,

676

options.timeout is not None,

677

options.extended_timeout is not None,

678

options.interval is not None,

679

options.approved_by_default is not None,

680

options.approval_delay is not None,

681

options.approval_duration is not None,

682

options.host is not None,

683

options.secret is not None,

684

options.approve,

685

options.deny))

673

686

674

687

if has_actions(options) and not (options.client or options.all):

675

688

parser.error("Options require clients names or --all.")

683

696

if options.is_enabled and len(options.client) > 1:

684

697

parser.error("--is-enabled requires exactly one client")

685

698

699

700

def main():

701

parser = argparse.ArgumentParser()

702

703

add_command_line_options(parser)

704

705

options = parser.parse_args()

706

707

check_option_syntax(parser, options)

708

686

709

clientnames = options.client

687

710

711

if options.debug:

712

log.setLevel(logging.DEBUG)

713

688

714

try:

689

715

bus = dbus.SystemBus()

716

log.debug("D-Bus: Connect to: (name=%r, path=%r)", busname,

717

server_path)

690

718

mandos_dbus_objc = bus.get_object(busname, server_path)

691

719

except dbus.exceptions.DBusException:

692

720

log.critical("Could not connect to Mandos server")

705

733

dbus_filter = NullFilter()

706

734

try:

707

735

dbus_logger.addFilter(dbus_filter)

736

log.debug("D-Bus: %s:%s:%s.GetManagedObjects()", busname,

737

server_path, dbus.OBJECT_MANAGER_IFACE)

708

738

mandos_clients = {path: ifs_and_props[client_interface]

709

739

for path, ifs_and_props in

710

740

mandos_serv_object_manager

1093

1123

class TestSetSecretCmd(TestValueArgumentPropertyCmd):

1094

1124

command = SetSecretCmd

1095

1125

property = "Secret"

1096

values_to_set = [open("/dev/null", "rb"),

1126

values_to_set = [io.BytesIO(b""),

1097

1127

io.BytesIO(b"secret\0xyzzy\nbar")]

1098

1128

values_to_get = [b"", b"secret\0xyzzy\nbar"]

1099

1129

1100

1130

class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):

1101

1131

command = SetTimeoutCmd

1102

1132

property = "Timeout"

1103

values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]

1104

values_to_get = [0, 300000, 1000, 120000, 31449600000]

1133

values_to_set = [datetime.timedelta(),

1134

datetime.timedelta(minutes=5),

1135

datetime.timedelta(seconds=1),

1136

datetime.timedelta(weeks=1),

1137

datetime.timedelta(weeks=52)]

1138

values_to_get = [0, 300000, 1000, 604800000, 31449600000]

1105

1139

1106

1140

class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):

1107

1141

command = SetExtendedTimeoutCmd

1108

1142

property = "ExtendedTimeout"

1109

values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]

1110

values_to_get = [0, 300000, 1000, 120000, 31449600000]

1143

values_to_set = [datetime.timedelta(),

1144

datetime.timedelta(minutes=5),

1145

datetime.timedelta(seconds=1),

1146

datetime.timedelta(weeks=1),

1147

datetime.timedelta(weeks=52)]

1148

values_to_get = [0, 300000, 1000, 604800000, 31449600000]

1111

1149

1112

1150

class TestSetIntervalCmd(TestValueArgumentPropertyCmd):

1113

1151

command = SetIntervalCmd

1114

1152

property = "Interval"

1115

values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]

1116

values_to_get = [0, 300000, 1000, 120000, 31449600000]

1153

values_to_set = [datetime.timedelta(),

1154

datetime.timedelta(minutes=5),

1155

datetime.timedelta(seconds=1),

1156

datetime.timedelta(weeks=1),

1157

datetime.timedelta(weeks=52)]

1158

values_to_get = [0, 300000, 1000, 604800000, 31449600000]

1117

1159

1118

1160

class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):

1119

1161

command = SetApprovalDelayCmd

1120

1162

property = "ApprovalDelay"

1121

values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]

1122

values_to_get = [0, 300000, 1000, 120000, 31449600000]

1163

values_to_set = [datetime.timedelta(),

1164

datetime.timedelta(minutes=5),

1165

datetime.timedelta(seconds=1),

1166

datetime.timedelta(weeks=1),

1167

datetime.timedelta(weeks=52)]

1168

values_to_get = [0, 300000, 1000, 604800000, 31449600000]

1123

1169

1124

1170

class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):

1125

1171

command = SetApprovalDurationCmd

1126

1172

property = "ApprovalDuration"

1127

values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]

1128

values_to_get = [0, 300000, 1000, 120000, 31449600000]

1173

values_to_set = [datetime.timedelta(),

1174

datetime.timedelta(minutes=5),

1175

datetime.timedelta(seconds=1),

1176

datetime.timedelta(weeks=1),

1177

datetime.timedelta(weeks=52)]

1178

values_to_get = [0, 300000, 1000, 604800000, 31449600000]

1129

1179

1130

1180

class Test_command_from_options(unittest.TestCase):

1131

1181

def setUp(self):

1135

1185

"""Assert that parsing ARGS should result in an instance of

1136

1186

COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""

1137

1187

options = self.parser.parse_args(args)

1188

check_option_syntax(self.parser, options)

1138

1189

commands = commands_from_options(options)

1139

1190

self.assertEqual(len(commands), 1)

1140

1191

command = commands[0]

1149

1200

self.assert_command_from_args(["--verbose"], PrintTableCmd,

1150

1201

verbose=True)

1151

1202

1203

def test_print_table_verbose_short(self):

1204

self.assert_command_from_args(["-v"], PrintTableCmd,

1205

verbose=True)

1206

1152

1207

def test_enable(self):

1153

1208

self.assert_command_from_args(["--enable", "foo"], EnableCmd)

1154

1209

1210

def test_enable_short(self):

1211

self.assert_command_from_args(["-e", "foo"], EnableCmd)

1212

1155

1213

def test_disable(self):

1156

1214

self.assert_command_from_args(["--disable", "foo"],

1157

1215

DisableCmd)

1158

1216

1217

def test_disable_short(self):

1218

self.assert_command_from_args(["-d", "foo"], DisableCmd)

1219

1159

1220

def test_bump_timeout(self):

1160

1221

self.assert_command_from_args(["--bump-timeout", "foo"],

1161

1222

BumpTimeoutCmd)

1162

1223

1224

def test_bump_timeout_short(self):

1225

self.assert_command_from_args(["-b", "foo"], BumpTimeoutCmd)

1226

1163

1227

def test_start_checker(self):

1164

1228

self.assert_command_from_args(["--start-checker", "foo"],

1165

1229

StartCheckerCmd)

1172

1236

self.assert_command_from_args(["--remove", "foo"],

1173

1237

RemoveCmd)

1174

1238

1239

def test_remove_short(self):

1240

self.assert_command_from_args(["-r", "foo"], RemoveCmd)

1241

1175

1242

def test_checker(self):

1176

1243

self.assert_command_from_args(["--checker", ":", "foo"],

1177

1244

SetCheckerCmd, value_to_set=":")

1180

1247

self.assert_command_from_args(["--checker", "", "foo"],

1181

1248

SetCheckerCmd, value_to_set="")

1182

1249

1250

def test_checker_short(self):

1251

self.assert_command_from_args(["-c", ":", "foo"],

1252

SetCheckerCmd, value_to_set=":")

1253

1183

1254

def test_timeout(self):

1184

1255

self.assert_command_from_args(["--timeout", "PT5M", "foo"],

1185

1256

SetTimeoutCmd,

1186

1257

value_to_set=300000)

1187

1258

1259

def test_timeout_short(self):

1260

self.assert_command_from_args(["-t", "PT5M", "foo"],

1261

SetTimeoutCmd,

1262

value_to_set=300000)

1263

1188

1264

def test_extended_timeout(self):

1189

1265

self.assert_command_from_args(["--extended-timeout", "PT15M",

1190

1266

"foo"],

1196

1272

SetIntervalCmd,

1197

1273

value_to_set=120000)

1198

1274

1275

def test_interval_short(self):

1276

self.assert_command_from_args(["-i", "PT2M", "foo"],

1277

SetIntervalCmd,

1278

value_to_set=120000)

1279

1199

1280

def test_approve_by_default(self):

1200

1281

self.assert_command_from_args(["--approve-by-default", "foo"],

1201

1282

ApproveByDefaultCmd)

1219

1300

"foo"], SetHostCmd,

1220

1301

value_to_set="foo.example.org")

1221

1302

1303

def test_host_short(self):

1304

self.assert_command_from_args(["-H", "foo.example.org",

1305

"foo"], SetHostCmd,

1306

value_to_set="foo.example.org")

1307

1222

1308

def test_secret_devnull(self):

1223

1309

self.assert_command_from_args(["--secret", os.path.devnull,

1224

1310

"foo"], SetSecretCmd,

1233

1319

"foo"], SetSecretCmd,

1234

1320

value_to_set=value)

1235

1321

1322

def test_secret_devnull_short(self):

1323

self.assert_command_from_args(["-s", os.path.devnull, "foo"],

1324

SetSecretCmd, value_to_set=b"")

1325

1326

def test_secret_tempfile_short(self):

1327

with tempfile.NamedTemporaryFile(mode="r+b") as f:

1328

value = b"secret\0xyzzy\nbar"

1329

f.write(value)

1330

f.seek(0)

1331

self.assert_command_from_args(["-s", f.name, "foo"],

1332

SetSecretCmd,

1333

value_to_set=value)

1334

1236

1335

def test_approve(self):

1237

1336

self.assert_command_from_args(["--approve", "foo"],

1238

1337

ApproveCmd)

1239

1338

1339

def test_approve_short(self):

1340

self.assert_command_from_args(["-A", "foo"], ApproveCmd)

1341

1240

1342

def test_deny(self):

1241

1343

self.assert_command_from_args(["--deny", "foo"], DenyCmd)

1242

1344

1345

def test_deny_short(self):

1346

self.assert_command_from_args(["-D", "foo"], DenyCmd)

1347

1243

1348

def test_dump_json(self):

1244

1349

self.assert_command_from_args(["--dump-json"], DumpJSONCmd)

1245

1350

1247

1352

self.assert_command_from_args(["--is-enabled", "foo"],

1248

1353

IsEnabledCmd)

1249

1354

1355

def test_is_enabled_short(self):

1356

self.assert_command_from_args(["-V", "foo"], IsEnabledCmd)

1357

1358

def test_deny_before_remove(self):

1359

options = self.parser.parse_args(["--deny", "--remove", "foo"])

1360

check_option_syntax(self.parser, options)

1361

commands = commands_from_options(options)

1362

self.assertEqual(len(commands), 2)

1363

self.assertIsInstance(commands[0], DenyCmd)

1364

self.assertIsInstance(commands[1], RemoveCmd)

1365

1366

def test_deny_before_remove_reversed(self):

1367

options = self.parser.parse_args(["--remove", "--deny", "--all"])

1368

check_option_syntax(self.parser, options)

1369

commands = commands_from_options(options)

1370

self.assertEqual(len(commands), 2)

1371

self.assertIsInstance(commands[0], DenyCmd)

1372

self.assertIsInstance(commands[1], RemoveCmd)

1373

1374

1375

class Test_check_option_syntax(unittest.TestCase):

1376

# This mostly corresponds to the definition from has_actions() in

1377

# check_option_syntax()

1378

actions = {

1379

# The actual values set here are not that important, but we do

1380

# at least stick to the correct types, even though they are

1381

# never used

1382

"enable": True,

1383

"disable": True,

1384

"bump_timeout": True,

1385

"start_checker": True,

1386

"stop_checker": True,

1387

"is_enabled": True,

1388

"remove": True,

1389

"checker": "x",

1390

"timeout": datetime.timedelta(),

1391

"extended_timeout": datetime.timedelta(),

1392

"interval": datetime.timedelta(),

1393

"approved_by_default": True,

1394

"approval_delay": datetime.timedelta(),

1395

"approval_duration": datetime.timedelta(),

1396

"host": "x",

1397

"secret": io.BytesIO(b"x"),

1398

"approve": True,

1399

"deny": True,

1400

}

1401

1402

def setUp(self):

1403

self.parser = argparse.ArgumentParser()

1404

add_command_line_options(self.parser)

1405

1406

@contextlib.contextmanager

1407

def assertParseError(self):

1408

with self.assertRaises(SystemExit) as e:

1409

with self.temporarily_suppress_stderr():

1410

yield

1411

# Exit code from argparse is guaranteed to be "2". Reference:

1412

# https://docs.python.org/3/library/argparse.html#exiting-methods

1413

self.assertEqual(e.exception.code, 2)

1414

1415

@staticmethod

1416

@contextlib.contextmanager

1417

def temporarily_suppress_stderr():

1418

null = os.open(os.path.devnull, os.O_RDWR)

1419

stderrcopy = os.dup(sys.stderr.fileno())

1420

os.dup2(null, sys.stderr.fileno())

1421

os.close(null)

1422

try:

1423

yield

1424

finally:

1425

# restore stderr

1426

os.dup2(stderrcopy, sys.stderr.fileno())

1427

os.close(stderrcopy)

1428

1429

def check_option_syntax(self, options):

1430

check_option_syntax(self.parser, options)

1431

1432

def test_actions_requires_client_or_all(self):

1433

for action, value in self.actions.items():

1434

options = self.parser.parse_args()

1435

setattr(options, action, value)

1436

with self.assertParseError():

1437

self.check_option_syntax(options)

1438

1439

def test_actions_conflicts_with_verbose(self):

1440

for action, value in self.actions.items():

1441

options = self.parser.parse_args()

1442

setattr(options, action, value)

1443

options.verbose = True

1444

with self.assertParseError():

1445

self.check_option_syntax(options)

1446

1447

def test_dump_json_conflicts_with_verbose(self):

1448

options = self.parser.parse_args()

1449

options.dump_json = True

1450

options.verbose = True

1451

with self.assertParseError():

1452

self.check_option_syntax(options)

1453

1454

def test_dump_json_conflicts_with_action(self):

1455

for action, value in self.actions.items():

1456

options = self.parser.parse_args()

1457

setattr(options, action, value)

1458

options.dump_json = True

1459

with self.assertParseError():

1460

self.check_option_syntax(options)

1461

1462

def test_all_can_not_be_alone(self):

1463

options = self.parser.parse_args()

1464

options.all = True

1465

with self.assertParseError():

1466

self.check_option_syntax(options)

1467

1468

def test_all_is_ok_with_any_action(self):

1469

for action, value in self.actions.items():

1470

options = self.parser.parse_args()

1471

setattr(options, action, value)

1472

options.all = True

1473

self.check_option_syntax(options)

1474

1475

def test_is_enabled_fails_without_client(self):

1476

options = self.parser.parse_args()

1477

options.is_enabled = True

1478

with self.assertParseError():

1479

self.check_option_syntax(options)

1480

1481

def test_is_enabled_works_with_one_client(self):

1482

options = self.parser.parse_args()

1483

options.is_enabled = True

1484

options.client = ["foo"]

1485

self.check_option_syntax(options)

1486

1487

def test_is_enabled_fails_with_two_clients(self):

1488

options = self.parser.parse_args()

1489

options.is_enabled = True

1490

options.client = ["foo", "barbar"]

1491

with self.assertParseError():

1492

self.check_option_syntax(options)

1493

1250

1494

1251

1495

1252

1496

def should_only_run_tests():