Coverage for python/univention/config_registry/frontend.py: 99%

393 statements  

« prev     ^ index     » next       coverage.py v7.6.0, created at 2026-09-11 14:11 +0000

1"""Univention Configuration Registry command line implementation.""" 

2# main configuration registry classes 

3# 

4# SPDX-FileCopyrightText: 2004-2026 Univention GmbH 

5# SPDX-License-Identifier: AGPL-3.0-only 

6# 

7# API stability :pylint: disable-msg=W0613 

8# Rewrite :pylint: disable-msg=R0912,R0914,R0915 

9 

10import os 

11import re 

12import sys 

13import time 

14from collections.abc import Callable, Iterator 

15from typing import IO, Any, NoReturn 

16 

17import univention.config_registry_info as cri 

18from univention.config_registry.backend import SCOPE, ConfigRegistry, exception_occured 

19from univention.config_registry.filters import filter_keys_only, filter_shell, filter_sort 

20from univention.config_registry.handler import ConfigHandlers, run_filter 

21from univention.config_registry.misc import escape_value, validate_key 

22from univention.config_registry.validation import Type 

23 

24 

25__all__ = [ 

26 'REPLOG_FILE', 

27 'UnknownKeyException', 

28 'handler_commit', 

29 'handler_dump', 

30 'handler_filter', 

31 'handler_get', 

32 'handler_help', 

33 'handler_info', 

34 'handler_register', 

35 'handler_search', 

36 'handler_set', 

37 'handler_unregister', 

38 'handler_unset', 

39 'handler_update', 

40 'handler_version', 

41 'main', 

42] 

43 

44REPLOG_FILE = '/var/log/univention/config-registry.replog' 

45 

46_SHOW_EMPTY, _SHOW_DESCRIPTION, _SHOW_SCOPE, _SHOW_CATEGORIES, _SHOW_DEFAULT, _SHOW_TYPE = (1 << _ for _ in range(6)) 

47 

48 

49class UnknownKeyException(Exception): 

50 """Query for unknown key: no info file nor set.""" 

51 

52 def __init__(self, value): 

53 Exception.__init__(self, value) 

54 

55 def __str__(self): 

56 return 'W: Unknown key: "%s"' % self.args 

57 

58 

59def replog(ucr: ConfigRegistry, var: str, old_value: str | None, value: str | None = None) -> None: 

60 """ 

61 This function writes a new entry to replication logfile if 

62 this feature has been enabled. 

63 

64 :param ucr: UCR instance. 

65 :param var: UCR variable name. 

66 :param old_value: Old UCR variable value. 

67 :param value: New UCR variable value. `None` is now unset. 

68 """ 

69 if ucr.is_true('ucr/replog/enabled', False): 

70 if value is not None: 

71 method = 'set' 

72 varvalue = "%s=%s" % (var, escape_value(value)) 

73 else: 

74 method = 'unset' 

75 varvalue = "'%s'" % var 

76 

77 scope_arg = { 

78 ConfigRegistry.LDAP: '--ldap-policy ', 

79 ConfigRegistry.FORCED: '--force ', 

80 ConfigRegistry.SCHEDULE: '--schedule ', 

81 }.get(ucr.scope, '') 

82 

83 if old_value is None: 

84 old_value = "[Previously undefined]" 

85 

86 log = '%s: %s %s%s old:%s\n' % (time.strftime("%Y-%m-%d %H:%M:%S"), method, scope_arg, varvalue, old_value) 

87 try: 

88 if not os.path.isfile(REPLOG_FILE): 88 ↛ 91line 88 didn't jump to line 91 because the condition on line 88 was always true

89 os.close(os.open(REPLOG_FILE, os.O_CREAT, 0o640)) 

90 

91 with open(REPLOG_FILE, "a+", encoding='utf-8') as logfile: 

92 logfile.write(log) 

93 except OSError as ex: 

94 print("E: exception occurred while writing to replication log: %s" % (ex,), file=sys.stderr) 

95 exception_occured() 

96 

97 

98def handler_set(args: list[str], opts: dict[str, Any] = {}, quiet: bool = False) -> None: 

99 """ 

100 Set config registry variables in args. 

101 Args is an array of strings 'key=value' or 'key?value'. 

102 

103 :param args: Command line arguments. 

104 :param opts: Command line options. 

105 :param quiet: Hide output. 

106 """ 

107 ucr = ConfigRegistry() 

108 ucr.load() 

109 ignore_check = opts.get('ignore-check') or ucr.is_false('ucr/check/type') 

110 

111 ucr = _ucr_from_opts(opts) 

112 info = _get_config_registry_info() 

113 with ucr: 

114 changes: dict[str, str | None] = {} 

115 for arg in args: 

116 sep_set = arg.find('=') # set 

117 sep_def = arg.find('?') # set if not already set 

118 if sep_set == -1 and sep_def == -1: 

119 print("W: Missing value for config registry variable '%s'" % (arg,), file=sys.stderr) 

120 continue 

121 

122 if sep_set > 0 and sep_def == -1: 

123 sep = sep_set 

124 elif sep_def > 0 and sep_set == -1: 

125 sep = sep_def 

126 else: 

127 sep = min(sep_set, sep_def) 

128 key = arg[0:sep] 

129 value = arg[sep + 1:] 

130 key_exists = key in ucr._layer 

131 do_set_value = (not key_exists or sep == sep_set) 

132 if do_set_value and validate_key(key): 

133 if not quiet: 133 ↛ 138line 133 didn't jump to line 138 because the condition on line 133 was always true

134 if key_exists: 

135 print('Setting %s' % key) 

136 else: 

137 print('Create %s' % key) 

138 vinfo = info.get_variable(key) or info.match_pattern(key) 

139 if vinfo: # Type checking can only be done if key is already configured 

140 try: 

141 validator = Type(vinfo) 

142 except (TypeError, ValueError): 

143 if ignore_check: 

144 print('W: Invalid UCR type definition for type %r of %r, but set anyway' % (vinfo.get('type'), key), file=sys.stderr) 

145 else: 

146 print('E: Invalid UCR type definition for type %r of %r, value %r not set' % (vinfo.get('type'), key, value), file=sys.stderr) 

147 opts['exit_code'] = 1 

148 opts.setdefault('type_def_error', []).append((key, vinfo.get('type'), value)) 

149 continue # do not set value and continue with next element of for loop to be set 

150 else: 

151 if not validator.check(value): 

152 if ignore_check: 

153 print('W: Value %r incompatible for %r, but setting anyway' % (value, key), file=sys.stderr) 

154 else: 

155 print('E: Value %r incompatible for %r' % (value, key), file=sys.stderr) 

156 opts['exit_code'] = 2 

157 opts.setdefault('type_errors', []).append((key, value)) 

158 continue # do not set value and continue with next element of for loop to be set 

159 changes[key] = value 

160 else: 

161 if do_set_value: 

162 opts['exit_code'] = 1 

163 if not quiet: 163 ↛ 115line 163 didn't jump to line 115 because the condition on line 163 was always true

164 if key_exists: 

165 print('Not updating %s' % key) 

166 else: 

167 print('Not setting %s' % key) 

168 changed = ucr.update(changes) 

169 

170 _run_changed(ucr, changed, "" if quiet else 'W: %s is overridden by scope "%s"') 

171 

172 

173def handler_unset(args: list[str], opts: dict[str, Any] = {}) -> None: 

174 """ 

175 Unset config registry variables in args. 

176 

177 :param args: Command line arguments. 

178 :param opts: Command line options. 

179 """ 

180 ucr = _ucr_from_opts(opts) 

181 with ucr: 

182 changes: dict[str, str | None] = {} 

183 for arg in args: 

184 if arg in ucr._layer: 

185 print('Unsetting %s' % arg) 

186 changes[arg] = None 

187 else: 

188 msg = "W: The config registry variable '%s' does not exist" 

189 print(msg % (arg,), file=sys.stderr) 

190 changed = ucr.update(changes) 

191 

192 _run_changed(ucr, changed, 'W: %s is still set in scope "%s"') 

193 

194 

195def ucr_update(ucr: ConfigRegistry, changes: dict[str, str | None]) -> None: 

196 """ 

197 Set or unset the given config registry variables. 

198 

199 :param ucr: UCR instance. 

200 :param changes: Changed UCR variables. 

201 """ 

202 with ucr: 

203 changed = ucr.update(changes) 

204 _run_changed(ucr, changed) 

205 

206 

207def _run_changed(ucr: ConfigRegistry, changed: dict[str, tuple[str | None, str | None]], msg: str = "") -> None: 

208 """ 

209 Run handlers for changed UCR variables. 

210 

211 :param ucr: UCR instance. 

212 :param changed: Mapping from UCR variable name to 2-tuple (old-value, new-value). 

213 :param msg: Message to be printed when change is shadowed by higher layer. Must contain 2 `%s` placeholders for `key` and `scope-name`. 

214 """ 

215 visible: dict[str, tuple[str | None, str | None]] = {} 

216 for key, (old_value, new_value) in changed.items(): 

217 replog(ucr, key, old_value, new_value) 

218 

219 reg = ucr.scope 

220 while old_value is None and reg > 0: 

221 reg -= 1 

222 old_value = ucr._registry[reg].get(key) 

223 

224 reg, new_value = ucr.get(key, (0, None), getscope=True) 

225 if reg > ucr.scope: 

226 if msg: 

227 print(msg % (key, SCOPE[reg]), file=sys.stderr) 

228 else: 

229 visible[key] = (old_value, new_value) 

230 

231 handlers = ConfigHandlers() 

232 handlers.load() 

233 handlers(list(visible), (ucr, visible)) 

234 

235 

236def _ucr_from_opts(opts: dict[str, Any]) -> ConfigRegistry: 

237 """ 

238 Create :py:class:`ConfigRegistry` instance according to requested layer. 

239 

240 :param opts: Command line options. 

241 :returns: A new UCR instance. 

242 """ 

243 if opts.get('ldap-policy'): 

244 scope = ConfigRegistry.LDAP 

245 elif opts.get('force'): 

246 scope = ConfigRegistry.FORCED 

247 elif opts.get('schedule'): 

248 scope = ConfigRegistry.SCHEDULE 

249 else: 

250 scope = ConfigRegistry.NORMAL 

251 ucr = ConfigRegistry(write_registry=scope) 

252 return ucr 

253 

254 

255def handler_dump(args: list[str], opts: dict[str, Any] = {}) -> Iterator[str]: 

256 """ 

257 Dump all variables. 

258 

259 :param args: Command line arguments. 

260 :param opts: Command line options. 

261 """ 

262 ucr = ConfigRegistry() 

263 ucr.load() 

264 yield from str(ucr).split('\n') 

265 

266 

267def handler_update(args: list[str], opts: dict[str, Any] = {}) -> None: 

268 """ 

269 Update handlers. 

270 

271 :param args: Command line arguments. 

272 :param opts: Command line options. 

273 """ 

274 handlers = ConfigHandlers() 

275 cur = handlers.update() 

276 handlers.update_divert(cur) 

277 

278 ucr = ConfigRegistry() 

279 _register_variable_default_values(ucr) 

280 

281 

282def handler_commit(args: list[str], opts: dict[str, Any] = {}) -> None: 

283 """ 

284 Commit all registered templated files. 

285 

286 :param args: Command line arguments. 

287 :param opts: Command line options. 

288 """ 

289 ucr = ConfigRegistry() 

290 ucr.load() 

291 

292 handlers = ConfigHandlers() 

293 handlers.load() 

294 handlers.commit(ucr, args) 

295 

296 

297def handler_register(args: list[str], opts: dict[str, Any] = {}) -> None: 

298 """ 

299 Register new `.info` file. 

300 

301 :param args: Command line arguments. 

302 :param opts: Command line options. 

303 """ 

304 ucr = ConfigRegistry() 

305 ucr.load() 

306 

307 handlers = ConfigHandlers() 

308 handlers.update() # cache must be current 

309 # Bug #21263: by forcing an update here, the new .info file is already 

310 # incorporated. Calling register for multifiles will increment the 

311 # def_count a second time, which is not nice, but uncritical, since the 

312 # diversion is (re-)done when >= 1. 

313 

314 _register_variable_default_values(ucr) 

315 handlers.register(args[0], ucr) 

316 # handlers.commit((ucr, {})) 

317 

318 

319def handler_unregister(args: list[str], opts: dict[str, Any] = {}) -> None: 

320 """ 

321 Unregister old `.info` file. 

322 

323 :param args: Command line arguments. 

324 :param opts: Command line options. 

325 """ 

326 ucr = ConfigRegistry() 

327 ucr.load() 

328 

329 handlers = ConfigHandlers() 

330 cur = handlers.update() # cache must be current 

331 obsolete = handlers.unregister(args[0], ucr) 

332 handlers.update_divert(cur - obsolete) 

333 _register_variable_default_values(ucr) 

334 

335 

336def handler_filter(args: list[str], opts: dict[str, Any] = {}) -> None: 

337 """Run filter on STDIN to STDOUT.""" 

338 ucr = ConfigRegistry() 

339 ucr.load() 

340 sys.stdout.buffer.write(run_filter(sys.stdin.read(), ucr, opts=opts)) 

341 

342 

343def handler_search(args: list[str], opts: dict[str, Any] = {}) -> Iterator[str]: 

344 """ 

345 Search for registry variable. 

346 

347 :param args: Command line arguments. 

348 :param opts: Command line options. 

349 """ 

350 search_keys = opts.get('key', False) 

351 search_values = opts.get('value', False) 

352 search_all = opts.get('all', False) 

353 count_search = int(search_keys) + int(search_values) + int(search_all) 

354 if count_search > 1: 

355 print('E: at most one out of [--key|--value|--all] may be set', file=sys.stderr) 

356 sys.exit(1) 

357 elif count_search == 0: 

358 search_keys = True 

359 search_values |= search_all 

360 search_keys |= search_all 

361 

362 if args: 

363 try: 

364 search = re.compile('|'.join('(?:%s)' % (_,) for _ in args)).search 

365 except re.error as ex: 

366 print('E: invalid regular expression: %s' % (ex,), file=sys.stderr) 

367 sys.exit(1) 

368 else: 

369 def search(x): # type: ignore[misc] 

370 return True 

371 

372 info = _get_config_registry_info() 

373 

374 category = opts.get('category') 

375 if category and not info.get_category(category): 

376 print('E: unknown category: "%s"' % (category,), file=sys.stderr) 

377 sys.exit(1) 

378 

379 ucr = ConfigRegistry() 

380 ucr.load() 

381 

382 details = _SHOW_EMPTY | _SHOW_DESCRIPTION 

383 if opts.get('non-empty'): 

384 details &= ~_SHOW_EMPTY 

385 if opts.get('brief') or ucr.is_true('ucr/output/brief', False): 

386 details &= ~_SHOW_DESCRIPTION 

387 if ucr.is_true('ucr/output/scope', False): 

388 details |= _SHOW_SCOPE 

389 if opts.get('verbose'): 

390 details |= _SHOW_CATEGORIES | _SHOW_DESCRIPTION 

391 

392 all_vars: dict[str, tuple[str | None, cri.Variable | None, int | None]] = {} # key: (value, vinfo, scope) 

393 for key, var in info.get_variables(category).items(): 

394 all_vars[key] = (None, var, None) 

395 for key, (scope, value) in ucr.items(getscope=True): 

396 try: 

397 all_vars[key] = (value, all_vars[key][1], scope) 

398 except LookupError: 

399 all_vars[key] = (value, None, scope) 

400 

401 for key, (value2, vinfo, _scope2) in all_vars.items(): 

402 if any(( 

403 search_keys and search(key), 

404 search_values and value2 and search(value2), 

405 search_all and vinfo and search(vinfo.get('description', '')), 

406 )): 

407 yield variable_info_string(key, value2, vinfo, details=details) 

408 

409 if _SHOW_EMPTY & details and not OPT_FILTERS['shell'][2]: 

410 patterns: dict = {} 

411 for arg in args or ('',): 

412 patterns.update(info.describe_search_term(arg)) 

413 for pattern, vinfo in patterns.items(): 

414 yield variable_info_string(pattern, None, vinfo, details=details) 

415 

416 

417def handler_get(args: list[str], opts: dict[str, Any] = {}) -> Iterator[str]: 

418 """ 

419 Return config registry variable. 

420 

421 :param args: Command line arguments. 

422 :param opts: Command line options. 

423 """ 

424 ucr = ConfigRegistry() 

425 ucr.load() 

426 key = args[0] 

427 value = ucr.get(key) 

428 if value is None: 

429 return 

430 elif OPT_FILTERS['shell'][2]: 

431 yield '%s: %s' % (key, value) 

432 else: 

433 yield value 

434 

435 

436def variable_info_string(key: str, value: str | None, variable_info: cri.Variable | None, scope: int | None = None, details: int = _SHOW_DESCRIPTION) -> str: 

437 """ 

438 Format UCR variable key, value, description, scope, categories and default value. 

439 

440 :param key: UCR variable name. 

441 :param value: UCR variable value. 

442 :param variable_info: Description object. 

443 :param scope: UCS layer. 

444 :param details: bit-field for detail-level. 

445 :returns: formatted string 

446 """ 

447 if value is None and not variable_info: 

448 raise UnknownKeyException(key) 

449 elif value in (None, '') and not _SHOW_EMPTY & details: 

450 return '' 

451 elif value is None: 

452 # if not shell filter option is set 

453 value_string = "<empty>" if not OPT_FILTERS["shell"][2] else "" 

454 else: 

455 value_string = '%s' % value 

456 

457 if scope is None or not 0 <= scope < len(SCOPE) or not _SHOW_SCOPE & details or OPT_FILTERS['shell'][2]: # Do not display scope in shell export 

458 key_value = '%s: %s' % (key, value_string) 

459 else: 

460 key_value = '%s (%s): %s' % (key, SCOPE[scope], value_string) 

461 

462 info = [key_value] 

463 if variable_info and _SHOW_DESCRIPTION & details: 

464 # info.append(' ' + variable_info.get('description', 

465 # 'no description available')) 

466 # <https://forge.univention.org/bugzilla/show_bug.cgi?id=15556> 

467 # Workaround: 

468 description = variable_info.get('description') 

469 if not description or not description.strip(): 

470 description = 'no description available' 

471 info.append(' ' + description) 

472 

473 if variable_info and _SHOW_CATEGORIES & details: 

474 info.append(' Categories: ' + variable_info.get('categories', 'none')) 

475 

476 if variable_info and _SHOW_DEFAULT & details: 

477 info.append(' Default: ' + variable_info.get('default', '(not set)')) 

478 

479 if variable_info and _SHOW_TYPE & details: 

480 try: 

481 validator = Type(variable_info) 

482 info.append(' Type: %s' % (validator,)) 

483 except ValueError as exc: 

484 info.append(' Type: %s' % exc) 

485 

486 if (_SHOW_CATEGORIES | _SHOW_DESCRIPTION) & details: 

487 info.append('') 

488 

489 return '\n'.join(info) 

490 

491 

492def handler_info(args: list[str], opts: dict[str, Any] = {}) -> Iterator[str]: 

493 """ 

494 Print variable info. 

495 

496 :param args: Command line arguments. 

497 :param opts: Command line options. 

498 """ 

499 ucr = ConfigRegistry() 

500 ucr.load() 

501 info = _get_config_registry_info() 

502 

503 for arg in args: 

504 try: 

505 yield variable_info_string( 

506 arg, ucr.get(arg, None), 

507 info.get_variable(arg), 

508 details=_SHOW_EMPTY | _SHOW_DESCRIPTION | _SHOW_CATEGORIES | _SHOW_DEFAULT | _SHOW_TYPE) 

509 except UnknownKeyException as ex: 

510 print(ex, file=sys.stderr) 

511 

512 

513def handler_version(args: list[str], opts: dict[str, Any] = {}) -> NoReturn: 

514 """ 

515 Print version info. 

516 

517 :param args: Command line arguments. 

518 :param opts: Command line options. 

519 """ 

520 print('univention-config-registry @%@package_version@%@') 

521 sys.exit(0) 

522 

523 

524def handler_help(args: list[str], opts: dict[str, Any] = {}, out: IO = sys.stdout) -> None: 

525 """ 

526 Print config registry command line usage. 

527 

528 :param args: Command line arguments. 

529 :param opts: Command line options. 

530 """ 

531 print(''' 

532univention-config-registry: base configuration for UCS 

533copyright (c) 2001-2026 Univention GmbH, Germany 

534 

535Syntax: 

536 univention-config-registry [options] <action> [options] [parameters] 

537 

538Options: 

539 

540 -h | --help | -?: 

541 print this usage message and exit program 

542 

543 --version | -v: 

544 print version information and exit program 

545 

546 --shell (valid actions: dump, search): 

547 convert key/value pair into shell compatible format, e.g. 

548 `version/version: 1.0` => `version_version="1.0"` 

549 

550 --keys-only (valid actions: dump, search): 

551 print only the keys 

552 

553Actions: 

554 set [--force|--schedule|--ldap-policy] [--ignore-check] <key>=<value> [... <key>=<value>]: 

555 set one or more keys to specified values; if a key is non-existent 

556 in the configuration registry it will be created 

557 --ignore-check: ignore check if given value is compatible with type of the key 

558 

559 get <key>: 

560 retrieve the value of the specified key from the configuration 

561 database 

562 

563 unset [--force|--schedule|--ldap-policy] <key> [... <key>]: 

564 remove one or more keys (and its associated values) from 

565 configuration database 

566 

567 dump: 

568 display all key/value pairs which are stored in the 

569 configuration database 

570 

571 search [--key|--value|--all] [--category <category>] [--brief|-verbose] \\ 

572 [--non-empty] [... <regex>]: 

573 displays all key/value pairs and their descriptions that match at 

574 least one of the given regular expressions 

575 --key: only search the keys (default) 

576 --value: only search the values 

577 --all: search keys, values and descriptions 

578 --category: limit search to variables of <category> 

579 --brief: don't print descriptions (default controlled via ucr/output/brief) 

580 --verbose: also print category for each variable 

581 --non-empty: only search in non-empty variables 

582 no <regex> given: display all variables 

583 

584 info <key> [... <key>]: 

585 display verbose information for the specified variable(s) 

586 

587 shell [key]: 

588 convert key/value pair into shell compatible format, e.g. 

589 `version/version: 1.0` => `version_version="1.0"` 

590 (deprecated: use --shell dump instead) 

591 

592 commit [file1 ...]: 

593 rebuild configuration file from univention template; if 

594 no file is specified ALL configuration files are rebuilt 

595 

596 filter [file]: 

597 evaluate a template file, expects Python inline code in UTF-8 or US-ASCII 

598 

599Description: 

600 univention-config-registry is a tool to handle the basic configuration for 

601 Univention Corporate Server (UCS) 

602''', file=out) 

603 sys.exit(0) 

604 

605 

606def missing_parameter(action: str) -> NoReturn: 

607 """Print missing parameter error.""" 

608 print('E: too few arguments for command [%s]' % (action,), file=sys.stderr) 

609 print('try `univention-config-registry --help` for more information', file=sys.stderr) 

610 sys.exit(1) 

611 

612 

613def _get_config_registry_info() -> cri.ConfigRegistryInfo: 

614 cri.set_language('en') 

615 return cri.ConfigRegistryInfo(install_mode=False) 

616 

617 

618def _register_variable_default_values(ucr: ConfigRegistry) -> None: 

619 """Create base-default.conf layer containig all default values""" 

620 info = _get_config_registry_info() 

621 _ucr = ConfigRegistry(write_registry=ConfigRegistry.DEFAULTS) 

622 _ucr.load() 

623 defaults: dict[str, str | None] = {} 

624 default_variables = info.get_variables() 

625 for key, variable in default_variables.items(): 

626 value = variable.get('default') 

627 if value: 627 ↛ 625line 627 didn't jump to line 625 because the condition on line 627 was always true

628 defaults[key] = value 

629 for key in _ucr: 

630 if key not in default_variables: 630 ↛ 629line 630 didn't jump to line 629 because the condition on line 630 was always true

631 defaults[key] = None 

632 

633 changed = {key: (old, new) for key, (old, new) in _ucr.update(defaults).items() if old != new} 

634 _ucr.save() 

635 ucr.load() 

636 _run_changed(ucr, changed, 'I: %s will be set in scope "%s"') 

637 

638 

639HANDLERS: dict[str, tuple[Callable[[list[str], dict[str, Any]], Iterator[str] | None], int]] = { 

640 'set': (handler_set, 1), 

641 'unset': (handler_unset, 1), 

642 'dump': (handler_dump, 0), 

643 'update': (handler_update, 0), 

644 'commit': (handler_commit, 0), 

645 'register': (handler_register, 1), 

646 'unregister': (handler_unregister, 1), 

647 'filter': (handler_filter, 0), 

648 'search': (handler_search, 0), 

649 'get': (handler_get, 1), 

650 'info': (handler_info, 1), 

651} 

652 

653# action options: each of these options perform an action 

654OPT_ACTIONS: dict[str, list] = { 

655 # name: [function, state, (alias list)] 

656 'help': [handler_help, False, ('-h', '-?')], 

657 'version': [handler_version, False, ('-v',)], 

658 'debug': [lambda args: None, False, ()], 

659} 

660 

661# filter options: these options define filter for the output 

662OPT_FILTERS: dict[str, list] = { 

663 # name: [prio, function, state, (valid actions)] 

664 'keys-only': [0, filter_keys_only, False, ('dump', 'search')], 

665 'sort': [10, filter_sort, False, ('dump', 'search', 'info')], 

666 'shell': [99, filter_shell, False, ('dump', 'search', 'get')], 

667} 

668 

669BOOL, STRING = range(2) 

670 

671OPT_COMMANDS: dict[str, dict[str, Any]] = { 

672 'set': { 

673 'force': [BOOL, False], 

674 'ldap-policy': [BOOL, False], 

675 'schedule': [BOOL, False], 

676 'ignore-check': [BOOL, False], 

677 }, 

678 'unset': { 

679 'force': [BOOL, False], 

680 'ldap-policy': [BOOL, False], 

681 'schedule': [BOOL, False], 

682 }, 

683 'search': { 

684 'key': [BOOL, False], 

685 'value': [BOOL, False], 

686 'all': [BOOL, False], 

687 'brief': [BOOL, False], 

688 'category': [STRING, None], 

689 'non-empty': [BOOL, False], 

690 'verbose': [BOOL, False], 

691 }, 

692 'filter': { 

693 'encode-utf8': [BOOL, False], 

694 'disallow-execution': [BOOL, False], 

695 }, 

696} 

697 

698 

699def main(args: list[str]) -> int: 

700 """Run config registry.""" 

701 try: 

702 # close your eyes ... 

703 if not args: 

704 args.append('--help') 

705 # search for options in command line arguments 

706 while args and args[0].startswith('-'): 

707 arg = args.pop(0) 

708 # is action option? 

709 for key, opt in OPT_ACTIONS.items(): 

710 if arg[2:] == key or arg in opt[2]: 

711 opt[1] = True 

712 break 

713 else: 

714 # not an action option; is a filter option? 

715 try: 

716 OPT_FILTERS[arg[2:]][2] = True 

717 except LookupError: 

718 print('E: unknown option %s' % (arg,), file=sys.stderr) 

719 sys.exit(1) 

720 

721 # is action already defined by global option? 

722 for name, (func, state, _aliases) in OPT_ACTIONS.items(): 

723 if state: 

724 func(args) 

725 

726 # find action 

727 try: 

728 action = args.pop(0) 

729 except IndexError: 

730 print('E: missing action, see --help', file=sys.stderr) 

731 sys.exit(1) 

732 # COMPAT: the 'shell' command is now an option and equivalent to 

733 # --shell search 

734 if action == 'shell': 

735 action = 'search' 

736 # activate shell option 

737 OPT_FILTERS['shell'][2] = True 

738 # switch to old, brief output 

739 OPT_COMMANDS['search']['brief'][1] = True 

740 

741 tmp = [] 

742 if not args: 

743 tmp.append('') 

744 else: 

745 for arg in args: 

746 if not arg.startswith('--'): 

747 tmp.append('^%s$' % arg) 

748 else: 

749 tmp.append(arg) 

750 args = tmp 

751 

752 # set 'sort' option by default for dump and search 

753 if action in ['dump', 'search', 'info']: 

754 OPT_FILTERS['sort'][2] = True 

755 

756 # set brief option when generating shell output 

757 if OPT_FILTERS['shell'][2]: 

758 OPT_COMMANDS['search']['brief'][1] = True 

759 

760 # if a filter option is set: verify that a valid command is given 

761 for name, (_prio, func, state, actions) in OPT_FILTERS.items(): 

762 if state and action not in actions: 

763 print('E: invalid option --%s for command %s' % (name, action), file=sys.stderr) 

764 sys.exit(1) 

765 

766 # check command options 

767 cmd_opts = OPT_COMMANDS.get(action, {}) 

768 while args and args[0].startswith('--'): 

769 arg = args.pop(0) 

770 if action in ('set', 'unset') and arg == '--forced': 

771 arg = '--force' 

772 try: 

773 cmd_opt_tuple = cmd_opts[arg[2:]] 

774 except LookupError: 

775 print('E: invalid option %s for command %s' % (arg, action), file=sys.stderr) 

776 sys.exit(1) 

777 else: 

778 if cmd_opt_tuple[0] == BOOL: 

779 cmd_opt_tuple[1] = True 

780 else: # STRING 

781 try: 

782 cmd_opt_tuple[1] = args.pop(0) 

783 except IndexError: 

784 msg = 'E: option %s for command %s expects an argument' 

785 print(msg % (arg, action), file=sys.stderr) 

786 sys.exit(1) 

787 

788 # Drop type 

789 cmd_opts = {key: value for key, (typ, value) in cmd_opts.items()} 

790 

791 # action! 

792 try: 

793 handler_func, min_args = HANDLERS[action] 

794 except LookupError: 

795 print('E: unknown action "%s", see --help' % (action,), file=sys.stderr) 

796 sys.exit(1) 

797 else: 

798 # enough arguments? 

799 if len(args) < min_args: 

800 missing_parameter(action) 

801 

802 # if any filter option is set 

803 cmd_opts['exit_code'] = 0 

804 result = handler_func(args, cmd_opts) 

805 if result is None: 

806 return cmd_opts['exit_code'] 

807 

808 results = result 

809 # let the filter options do their job 

810 for (_prio, filter_func, state, _actions) in sorted(OPT_FILTERS.values()): 

811 if not state: 

812 continue 

813 results = filter_func(args, results) 

814 

815 for line in results: 

816 print(line) 

817 

818 except (OSError, TypeError): 

819 if OPT_ACTIONS['debug'][1]: 

820 raise 

821 exception_occured() 

822 return cmd_opts['exit_code']