(self, params, target=None, **kwargs)
| 304 | return scim_push_parser |
| 305 | |
| 306 | def execute(self, params, target=None, **kwargs): |
| 307 | scim = find_scim(params, target) |
| 308 | dry_run = kwargs.get('dry_run') is True |
| 309 | |
| 310 | scim_url = get_scim_url(params, scim['node_id']) |
| 311 | record_uid = kwargs.get('record') |
| 312 | record = None # type: Optional[vault.TypedRecord] |
| 313 | if record_uid: |
| 314 | if record_uid in params.record_cache: |
| 315 | r = vault.KeeperRecord.load(params, record_uid) |
| 316 | if isinstance(r, vault.TypedRecord): |
| 317 | record = r |
| 318 | else: |
| 319 | raise CommandError('', f'Record UID "{record_uid}": invalid record type ') |
| 320 | else: |
| 321 | raise CommandError('', f'Record UID "{record_uid}": does not exist') |
| 322 | else: |
| 323 | for r in vault_extensions.find_records(params, record_version=3): |
| 324 | if not isinstance(r, vault.TypedRecord): |
| 325 | continue |
| 326 | field = next((x for x in r.fields if x.type == 'url'), None) |
| 327 | if field: |
| 328 | url = field.get_default_value(str) |
| 329 | if url and url == scim_url: |
| 330 | record = r |
| 331 | break |
| 332 | if not isinstance(record, vault.TypedRecord): |
| 333 | raise CommandError('', f'Cannot find SCIM record with URL: {scim_url}') |
| 334 | |
| 335 | destructive = 0 |
| 336 | verbose = False |
| 337 | field = next((x for x in record.fields if x.type == 'password'), None) |
| 338 | if not field: |
| 339 | raise CommandError('', f'"Password" field not found on record "{record.title}"') |
| 340 | token = field.get_default_value(str) |
| 341 | if not token: |
| 342 | raise CommandError('', f'"Password" field is empty on record "{record.title}"') |
| 343 | |
| 344 | field = record.get_typed_field('text', 'Destructive') |
| 345 | if field: |
| 346 | sv = field.get_default_value(str) |
| 347 | if sv and sv.isnumeric(): |
| 348 | destructive = int(sv) |
| 349 | field = record.get_typed_field('text', 'Verbose') |
| 350 | if field: |
| 351 | sv = field.get_default_value(str) |
| 352 | if sv and sv.lower() in ('true', 'yes', 'on', '1'): |
| 353 | verbose = True |
| 354 | |
| 355 | def verbose_logging(message): |
| 356 | if verbose: |
| 357 | logging.info(message) |
| 358 | |
| 359 | keeper_users = {} # type: Dict[str, ScimUser] |
| 360 | keeper_groups = {} # type: Dict[str, ScimGroup] |
| 361 | logging.debug('SCIM Query Keeper') |
| 362 | for element in ScimPushCommand.scim_keeper(scim_url, token): |
| 363 | if isinstance(element, ScimUser): |
nothing calls this directly
no test coverage detected