/mandos/release : revision 237.7.593

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.")

682

695

parser.error("--all requires an action.")

683

696

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

684

697

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

698

if options.remove:

699

options.remove = False

700

if has_actions(options) and not options.deny:

701

parser.error("--remove can only be combined with --deny")

702

options.remove = True

703

704

705

def main():

706

parser = argparse.ArgumentParser()

707

708

add_command_line_options(parser)

709

710

options = parser.parse_args()

711

712

check_option_syntax(parser, options)

685

713

686

714

clientnames = options.client

687

715

716

if options.debug:

717

log.setLevel(logging.DEBUG)

718

688

719

try:

689

720

bus = dbus.SystemBus()

721

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

722

server_path)

690

723

mandos_dbus_objc = bus.get_object(busname, server_path)

691

724

except dbus.exceptions.DBusException:

692

725

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

705

738

dbus_filter = NullFilter()

706

739

try:

707

740

dbus_logger.addFilter(dbus_filter)

741

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

742

server_path, dbus.OBJECT_MANAGER_IFACE)

708

743

mandos_clients = {path: ifs_and_props[client_interface]

709

744

for path, ifs_and_props in

710

745

mandos_serv_object_manager

1093

1128

class TestSetSecretCmd(TestValueArgumentPropertyCmd):

1094

1129

command = SetSecretCmd

1095

1130

property = "Secret"

1096

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

1131

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

1097

1132

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

1098

1133

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

1099

1134

1100

1135

class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):

1101

1136

command = SetTimeoutCmd

1102

1137

property = "Timeout"

1103

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

1104

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

1138

values_to_set = [datetime.timedelta(),

1139

datetime.timedelta(minutes=5),

1140

datetime.timedelta(seconds=1),

1141

datetime.timedelta(weeks=1),

1142

datetime.timedelta(weeks=52)]

1143

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

1105

1144

1106

1145

class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):

1107

1146

command = SetExtendedTimeoutCmd

1108

1147

property = "ExtendedTimeout"

1109

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

1110

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

1148

values_to_set = [datetime.timedelta(),

1149

datetime.timedelta(minutes=5),

1150

datetime.timedelta(seconds=1),

1151

datetime.timedelta(weeks=1),

1152

datetime.timedelta(weeks=52)]

1153

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

1111

1154

1112

1155

class TestSetIntervalCmd(TestValueArgumentPropertyCmd):

1113

1156

command = SetIntervalCmd

1114

1157

property = "Interval"

1115

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

1116

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

1158

values_to_set = [datetime.timedelta(),

1159

datetime.timedelta(minutes=5),

1160

datetime.timedelta(seconds=1),

1161

datetime.timedelta(weeks=1),

1162

datetime.timedelta(weeks=52)]

1163

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

1117

1164

1118

1165

class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):

1119

1166

command = SetApprovalDelayCmd

1120

1167

property = "ApprovalDelay"

1121

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

1122

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

1168

values_to_set = [datetime.timedelta(),

1169

datetime.timedelta(minutes=5),

1170

datetime.timedelta(seconds=1),

1171

datetime.timedelta(weeks=1),

1172

datetime.timedelta(weeks=52)]

1173

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

1123

1174

1124

1175

class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):

1125

1176

command = SetApprovalDurationCmd

1126

1177

property = "ApprovalDuration"

1127

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

1128

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

1178

values_to_set = [datetime.timedelta(),

1179

datetime.timedelta(minutes=5),

1180

datetime.timedelta(seconds=1),

1181

datetime.timedelta(weeks=1),

1182

datetime.timedelta(weeks=52)]

1183

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

1129

1184

1130

1185

class Test_command_from_options(unittest.TestCase):

1131

1186

def setUp(self):

1135

1190

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

1136

1191

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

1137

1192

options = self.parser.parse_args(args)

1193

check_option_syntax(self.parser, options)

1138

1194

commands = commands_from_options(options)

1139

1195

self.assertEqual(len(commands), 1)

1140

1196

command = commands[0]

1149

1205

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

1150

1206

verbose=True)

1151

1207

1208

def test_print_table_verbose_short(self):

1209

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

1210

verbose=True)

1211

1152

1212

def test_enable(self):

1153

1213

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

1154

1214

1215

def test_enable_short(self):

1216

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

1217

1155

1218

def test_disable(self):

1156

1219

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

1157

1220

DisableCmd)

1158

1221

1222

def test_disable_short(self):

1223

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

1224

1159

1225

def test_bump_timeout(self):

1160

1226

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

1161

1227

BumpTimeoutCmd)

1162

1228

1229

def test_bump_timeout_short(self):

1230

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

1231

1163

1232

def test_start_checker(self):

1164

1233

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

1165

1234

StartCheckerCmd)

1172

1241

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

1173

1242

RemoveCmd)

1174

1243

1244

def test_remove_short(self):

1245

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

1246

1175

1247

def test_checker(self):

1176

1248

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

1177

1249

SetCheckerCmd, value_to_set=":")

1178

1250

1251

def test_checker_empty(self):

1252

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

1253

SetCheckerCmd, value_to_set="")

1254

1255

def test_checker_short(self):

1256

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

1257

SetCheckerCmd, value_to_set=":")

1258

1179

1259

def test_timeout(self):

1180

1260

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

1181

1261

SetTimeoutCmd,

1182

1262

value_to_set=300000)

1183

1263

1264

def test_timeout_short(self):

1265

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

1266

SetTimeoutCmd,

1267

value_to_set=300000)

1268

1184

1269

def test_extended_timeout(self):

1185

1270

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

1186

1271

"foo"],

1192

1277

SetIntervalCmd,

1193

1278

value_to_set=120000)

1194

1279

1280

def test_interval_short(self):

1281

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

1282

SetIntervalCmd,

1283

value_to_set=120000)

1284

1195

1285

def test_approve_by_default(self):

1196

1286

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

1197

1287

ApproveByDefaultCmd)

1215

1305

"foo"], SetHostCmd,

1216

1306

value_to_set="foo.example.org")

1217

1307

1308

def test_host_short(self):

1309

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

1310

"foo"], SetHostCmd,

1311

value_to_set="foo.example.org")

1312

1218

1313

def test_secret_devnull(self):

1219

1314

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

1220

1315

"foo"], SetSecretCmd,

1229

1324

"foo"], SetSecretCmd,

1230

1325

value_to_set=value)

1231

1326

1327

def test_secret_devnull_short(self):

1328

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

1329

SetSecretCmd, value_to_set=b"")

1330

1331

def test_secret_tempfile_short(self):

1332

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

1333

value = b"secret\0xyzzy\nbar"

1334

f.write(value)

1335

f.seek(0)

1336

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

1337

SetSecretCmd,

1338

value_to_set=value)

1339

1232

1340

def test_approve(self):

1233

1341

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

1234

1342

ApproveCmd)

1235

1343

1344

def test_approve_short(self):

1345

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

1346

1236

1347

def test_deny(self):

1237

1348

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

1238

1349

1350

def test_deny_short(self):

1351

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

1352

1239

1353

def test_dump_json(self):

1240

1354

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

1241

1355

1243

1357

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

1244

1358

IsEnabledCmd)

1245

1359

1360

def test_is_enabled_short(self):

1361

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

1362

1363

def test_deny_before_remove(self):

1364

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

1365

check_option_syntax(self.parser, options)

1366

commands = commands_from_options(options)

1367

self.assertEqual(len(commands), 2)

1368

self.assertIsInstance(commands[0], DenyCmd)

1369

self.assertIsInstance(commands[1], RemoveCmd)

1370

1371

def test_deny_before_remove_reversed(self):

1372

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

1373

check_option_syntax(self.parser, options)

1374

commands = commands_from_options(options)

1375

self.assertEqual(len(commands), 2)

1376

self.assertIsInstance(commands[0], DenyCmd)

1377

self.assertIsInstance(commands[1], RemoveCmd)

1378

1379

1380

class Test_check_option_syntax(unittest.TestCase):

1381

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

1382

# check_option_syntax()

1383

actions = {

1384

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

1385

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

1386

# never used

1387

"enable": True,

1388

"disable": True,

1389

"bump_timeout": True,

1390

"start_checker": True,

1391

"stop_checker": True,

1392

"is_enabled": True,

1393

"remove": True,

1394

"checker": "x",

1395

"timeout": datetime.timedelta(),

1396

"extended_timeout": datetime.timedelta(),

1397

"interval": datetime.timedelta(),

1398

"approved_by_default": True,

1399

"approval_delay": datetime.timedelta(),

1400

"approval_duration": datetime.timedelta(),

1401

"host": "x",

1402

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

1403

"approve": True,

1404

"deny": True,

1405

}

1406

1407

def setUp(self):

1408

self.parser = argparse.ArgumentParser()

1409

add_command_line_options(self.parser)

1410

1411

@contextlib.contextmanager

1412

def assertParseError(self):

1413

with self.assertRaises(SystemExit) as e:

1414

with self.temporarily_suppress_stderr():

1415

yield

1416

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

1417

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

1418

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

1419

1420

@staticmethod

1421

@contextlib.contextmanager

1422

def temporarily_suppress_stderr():

1423

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

1424

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

1425

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

1426

os.close(null)

1427

try:

1428

yield

1429

finally:

1430

# restore stderr

1431

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

1432

os.close(stderrcopy)

1433

1434

def check_option_syntax(self, options):

1435

check_option_syntax(self.parser, options)

1436

1437

def test_actions_requires_client_or_all(self):

1438

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

1439

options = self.parser.parse_args()

1440

setattr(options, action, value)

1441

with self.assertParseError():

1442

self.check_option_syntax(options)

1443

1444

def test_actions_conflicts_with_verbose(self):

1445

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

1446

options = self.parser.parse_args()

1447

setattr(options, action, value)

1448

options.verbose = True

1449

with self.assertParseError():

1450

self.check_option_syntax(options)

1451

1452

def test_dump_json_conflicts_with_verbose(self):

1453

options = self.parser.parse_args()

1454

options.dump_json = True

1455

options.verbose = True

1456

with self.assertParseError():

1457

self.check_option_syntax(options)

1458

1459

def test_dump_json_conflicts_with_action(self):

1460

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

1461

options = self.parser.parse_args()

1462

setattr(options, action, value)

1463

options.dump_json = True

1464

with self.assertParseError():

1465

self.check_option_syntax(options)

1466

1467

def test_all_can_not_be_alone(self):

1468

options = self.parser.parse_args()

1469

options.all = True

1470

with self.assertParseError():

1471

self.check_option_syntax(options)

1472

1473

def test_all_is_ok_with_any_action(self):

1474

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

1475

options = self.parser.parse_args()

1476

setattr(options, action, value)

1477

options.all = True

1478

self.check_option_syntax(options)

1479

1480

def test_is_enabled_fails_without_client(self):

1481

options = self.parser.parse_args()

1482

options.is_enabled = True

1483

with self.assertParseError():

1484

self.check_option_syntax(options)

1485

1486

def test_is_enabled_works_with_one_client(self):

1487

options = self.parser.parse_args()

1488

options.is_enabled = True

1489

options.client = ["foo"]

1490

self.check_option_syntax(options)

1491

1492

def test_is_enabled_fails_with_two_clients(self):

1493

options = self.parser.parse_args()

1494

options.is_enabled = True

1495

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

1496

with self.assertParseError():

1497

self.check_option_syntax(options)

1498

1499

def test_remove_can_only_be_combined_with_action_deny(self):

1500

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

1501

if action in {"remove", "deny"}:

1502

continue

1503

options = self.parser.parse_args()

1504

setattr(options, action, value)

1505

options.all = True

1506

options.remove = True

1507

with self.assertParseError():

1508

self.check_option_syntax(options)

1509

1246

1510

1247

1511

1248

1512

def should_only_run_tests():