(self, params, **kwargs)
| 281 | return rotate_parser |
| 282 | |
| 283 | def execute(self, params, **kwargs): |
| 284 | print_result = kwargs['print'] if 'print' in kwargs else None |
| 285 | name = kwargs['name'] if 'name' in kwargs else None |
| 286 | match = kwargs['match'] if 'match' in kwargs else None |
| 287 | force = kwargs['force'] if 'force' in kwargs else None |
| 288 | if name: |
| 289 | record_uid = None |
| 290 | rotate_name = None |
| 291 | if name in params.record_cache: |
| 292 | record_uid = name |
| 293 | else: |
| 294 | RecordRotateCommand.find_endpoints(params) |
| 295 | nl = name.lower() |
| 296 | endpoints = [ |
| 297 | x for x in RecordRotateCommand.Endpoints if x.record_title.lower() == nl or x.name.lower() == nl |
| 298 | ] |
| 299 | if len(endpoints) > 0: |
| 300 | if len(endpoints) == 1: |
| 301 | record_uid = endpoints[0].record_uid |
| 302 | rotate_name = endpoints[0].name |
| 303 | else: |
| 304 | logging.error('There are more than one rotation records with name %s. Please use record UID.', name) |
| 305 | return |
| 306 | if record_uid: |
| 307 | rotate_password( |
| 308 | params, record_uid, rotate_name=rotate_name, plugin_name=kwargs.get('plugin'), |
| 309 | host=kwargs.get('host'), port=kwargs.get('port'), rules=kwargs.get('rules'), |
| 310 | length=kwargs.get('length'), new_password=kwargs.get('password') |
| 311 | ) |
| 312 | if print_result: |
| 313 | record = api.get_record(params, record_uid) |
| 314 | record.display() |
| 315 | else: |
| 316 | logging.error('Rotate {0}: not found'.format(name)) |
| 317 | elif match: |
| 318 | results = api.search_records(params, match) |
| 319 | for r in results: |
| 320 | if r.version not in (2, 3): |
| 321 | continue |
| 322 | if force or confirm(f'Rotate password for record {r.title}?'): |
| 323 | rotate_password( |
| 324 | params, r.record_uid, plugin_name=kwargs.get('plugin'), new_password=kwargs.get('password'), |
| 325 | host=kwargs.get('host'), port=kwargs.get('port'), rules=kwargs.get('rules') |
| 326 | ) |
| 327 | if print_result: |
| 328 | record = api.get_record(params, r.record_uid) |
| 329 | record.display() |
| 330 | else: |
| 331 | RecordRotateCommand.find_endpoints(params) |
| 332 | if RecordRotateCommand.Endpoints: |
| 333 | logging.info("Available records for password rotation") |
| 334 | logging.info('') |
| 335 | headers = ["#", 'Name', 'Type', 'Record UID', 'Record Title', 'Folder(s)'] |
| 336 | table = [] |
| 337 | for i in range(len(RecordRotateCommand.Endpoints)): |
| 338 | endpoint = RecordRotateCommand.Endpoints[i] |
| 339 | title = endpoint.record_title |
| 340 | folder = endpoint.paths[0] if len(endpoint.paths) > 0 else '/' |
no test coverage detected