Skip to content

Add NSF support for pam rotation script list/add/edit/delete #213

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mtyagi-ks wants to merge 1 commit into
base: pam-nsf-support-int
Choose a base branch
from
+259 −73
Open

Add NSF support for pam rotation script list/add/edit/delete #213

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
232 changes: 160 additions & 72 deletions keepercli-package/src/keepercli/commands/pam/pam_rotation.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,8 @@
'pamDatabase', 'pamDirectory', 'pamMachine', 'pamUser', 'pamRemoteBrowser',
)

_PAM_SCRIPT_RECORD_TYPES = ('pamUser', 'pamDirectory')


def _resolve_nsf_record_uid(vault: vault_online.VaultOnline, identifier: str) -> Optional[str]:
"""Resolve an NSF record UID from a UID or exact title."""
Expand Down Expand Up @@ -94,6 +96,130 @@ def _load_pam_typed_record(
return None


def _is_nsf_pam_record(vault: vault_online.VaultOnline, record_uid: str) -> bool:
return bool(record_uid and vault.nsf_data and vault.nsf_data.get_record(record_uid))


def _save_pam_typed_record(
vault: vault_online.VaultOnline, record: vault_record.TypedRecord) -> None:
"""Persist a PAM typed record via NSF or classic update (including file/script links)."""
_attach_record_key(vault, record)
if _is_nsf_pam_record(vault, record.record_uid):
nsf_management.update_nsf_typed_record(vault, record)
else:
record_management.update_record(vault, record)
vault.sync_requested = True


def _iter_nsf_pam_script_records(
vault: vault_online.VaultOnline, pattern: Optional[str] = None):
"""Yield NSF pamUser/pamDirectory records optionally filtered by UID/title pattern."""
if not vault.nsf_data:
return
pattern_cf = pattern.casefold() if pattern else None
for entry in vault.nsf_data.records():
typed = _load_nsf_typed_record(vault, entry.record_uid)
if not typed or typed.record_type not in _PAM_SCRIPT_RECORD_TYPES:
continue
if pattern_cf:
title = (typed.title or '').casefold()
if typed.record_uid != pattern and pattern_cf not in title and not fnmatch.fnmatch(title, pattern_cf):
continue
yield typed


def _find_pam_script_records(
vault: vault_online.VaultOnline, pattern: Optional[str] = None):
"""Find pamUser/pamDirectory TypedRecords in classic vault and NSF."""
found = []
seen = set()
for rec in vault.vault_data.find_records(
criteria=pattern, record_version=3, record_type=_PAM_SCRIPT_RECORD_TYPES):
loaded = vault.vault_data.load_record(rec.record_uid)
if not isinstance(loaded, vault_record.TypedRecord):
continue
found.append(loaded)
seen.add(loaded.record_uid)

if pattern:
nsf_uid = _resolve_nsf_record_uid(vault, pattern)
if nsf_uid and nsf_uid not in seen:
typed = _load_nsf_typed_record(vault, nsf_uid)
if typed and typed.record_type in _PAM_SCRIPT_RECORD_TYPES:
found.append(typed)
seen.add(typed.record_uid)
else:
for typed in _iter_nsf_pam_script_records(vault, pattern):
if typed.record_uid not in seen:
found.append(typed)
seen.add(typed.record_uid)
else:
for typed in _iter_nsf_pam_script_records(vault):
if typed.record_uid not in seen:
found.append(typed)
seen.add(typed.record_uid)
return found


def _get_unique_pam_script_record(
vault: vault_online.VaultOnline, record_name: str) -> vault_record.TypedRecord:
"""Resolve a single pamUser/pamDirectory for script commands (classic or NSF)."""
records = _find_pam_script_records(vault, record_name)
if len(records) == 0:
raise base.CommandError(f'Record "{record_name}" not found')
if len(records) > 1:
raise base.CommandError(f'Record "{record_name}" is not unique. Use record UID.')
return records[0]


def _load_script_file_record(vault: vault_online.VaultOnline, file_ref: str):
"""Load a script file attachment record from classic vault or NSF."""
if not file_ref:
return None
file_record = vault.vault_data.load_record(file_ref)
if file_record:
return file_record
return _load_nsf_typed_record(vault, file_ref)


def _find_script_value(vault, script_field, script_name):
"""Find a script value by fileRef UID or file title/name."""
if not script_field or not script_name:
return None
script_value = next(
(x for x in script_field.value if isinstance(x, dict) and x.get('fileRef') == script_name),
None)
if script_value is not None:
return script_value
s_name = script_name.casefold()
for x in script_field.value:
if not isinstance(x, dict):
continue
file_uid = x.get('fileRef')
file_record = _load_script_file_record(vault, file_uid)
if not file_record:
continue
if getattr(file_record, 'record_uid', None) == script_name:
return x
title = (getattr(file_record, 'title', None) or '').casefold()
name = (getattr(file_record, 'name', None) or '').casefold()
if title == s_name or name == s_name:
return x
return None


def _resolve_script_credential_uid(vault: vault_online.VaultOnline, ref: str) -> Optional[str]:
"""Resolve a credential UID for script recordRef (classic or NSF)."""
if not ref:
return None
loaded = _load_pam_typed_record(vault, ref)
if loaded:
return loaded.record_uid
if vault.vault_data.get_record_key(ref) or vault.vault_data.load_record(ref):
return ref
return None


def _iter_nsf_pam_configurations(vault: vault_online.VaultOnline):
"""Yield NSF PAM configuration TypedRecords."""
if not vault.nsf_data:
Expand Down Expand Up @@ -1275,25 +1401,22 @@ def execute(self, context: KeeperParams, **kwargs):

table = []
header = ['record_uid', 'title', 'record_type', 'script_uid', 'script_name', 'records', 'command']
for rec in vault.vault_data.find_records(criteria=pattern, record_version=3,
record_type=('pamUser', 'pamDirectory')):
record = vault.vault_data.load_record(rec.record_uid)
if not isinstance(record, vault_record.TypedRecord):
continue
for record in _find_pam_script_records(vault, pattern):
for field in (x for x in record.fields if x.type == 'script'):
value = field.get_default_value(dict)
if not value:
continue
file_ref = value.get('fileRef')
if not file_ref:
continue
file_record = vault.vault_data.load_record(file_ref)
if not file_record:
continue
records = value.get('recordRef')
command = value.get('command')
table.append([record.record_uid, record.title, record.record_type, file_record.record_uid,
file_record.title, records, command])
for value in (field.value or []):
if not isinstance(value, dict):
continue
file_ref = value.get('fileRef')
if not file_ref:
continue
file_record = _load_script_file_record(vault, file_ref)
if not file_record:
continue
records = value.get('recordRef')
command = value.get('command')
table.append([record.record_uid, record.title, record.record_type,
getattr(file_record, 'record_uid', file_ref),
getattr(file_record, 'title', file_ref), records, command])
fmt = kwargs.get('format')
if fmt != 'json':
header = [report_utils.field_to_title(x) for x in header]
Expand Down Expand Up @@ -1323,12 +1446,7 @@ def execute(self, context: KeeperParams, **kwargs):
record_name = kwargs.get('record')
if not record_name:
raise base.CommandError('"record" argument is required')
records = list(vault.vault_data.find_records(criteria=record_name, record_version=3, record_type=('pamUser', 'pamDirectory')))
if len(records) == 0:
raise base.CommandError(f'Record "{record_name}" not found')
if len(records) > 1:
raise base.CommandError(f'Record "{record_name}" is not unique. Use record UID.')
record = vault.vault_data.load_record(records[0].record_uid)
record = _get_unique_pam_script_record(vault, record_name)
if not isinstance(record, vault_record.TypedRecord):
raise base.CommandError(f'Record "{record.title}" is not a rotation record.')

Expand Down Expand Up @@ -1361,14 +1479,14 @@ def execute(self, context: KeeperParams, **kwargs):
record_refs = kwargs.get('add_credential')
if isinstance(record_refs, list):
for ref in record_refs:
if ref in vault.vault_data._records:
script_value['recordRef'].append(ref)
resolved = _resolve_script_credential_uid(vault, ref)
if resolved:
script_value['recordRef'].append(resolved)
cmd = kwargs.get('script_command')
if cmd:
script_value['command'] = cmd

record_management.update_record(vault, record)
vault.sync_data = True
_save_pam_typed_record(vault, record)


class PAMScriptEditCommand(base.ArgparseCommand):
Expand Down Expand Up @@ -1401,31 +1519,14 @@ def execute(self, context: KeeperParams, **kwargs):
if not script_name:
raise base.CommandError('"script" argument is required')

records = list(vault.vault_data.find_records(criteria=record_name, record_version=3, record_type=('pamUser', 'pamDirectory')))
if len(records) == 0:
raise base.CommandError(f'Record "{record_name}" not found')
if len(records) > 1:
raise base.CommandError(f'Record "{record_name}" is not unique. Use record UID.')
record = vault.vault_data.load_record(records[0].record_uid)
record = _get_unique_pam_script_record(vault, record_name)
if not isinstance(record, vault_record.TypedRecord):
raise base.CommandError(f'Record "{record.title}" is not a rotation record.')

script_field = next((x for x in record.fields if x.type == 'script'), None)
if script_field is None:
raise base.CommandError(f'Record "{record.title}" has no rotation scripts.')
script_value = next((x for x in script_field.value if x.get('fileRef') == script_name), None)
if script_value is None:
s_name = script_name.casefold()
for x in script_field.value:
file_uid = x.get('fileRef')
file_record = vault.vault_data.load_record(file_uid)
if isinstance(file_record, vault_record.FileRecord):
if file_record.record_uid == s_name:
script_value = x
break
elif file_record.title.casefold() == s_name:
script_value = x
break
script_value = _find_script_value(vault, script_field, script_name)

if not isinstance(script_value, dict):
raise base.CommandError(f'Record "{record.title}" does not have script "{script_name}"')
Expand All @@ -1437,11 +1538,17 @@ def execute(self, context: KeeperParams, **kwargs):
refs.update(record_refs)
remove_credential = kwargs.get('remove_credential')
if isinstance(remove_credential, list) and remove_credential:
refs.difference_update(remove_credential)
for ref in remove_credential:
resolved = _resolve_script_credential_uid(vault, ref) or ref
refs.discard(resolved)
refs.discard(ref)
modified = True
add_credential = kwargs.get('add_credential')
if isinstance(add_credential, list) and add_credential:
refs.update(add_credential)
for ref in add_credential:
resolved = _resolve_script_credential_uid(vault, ref)
if resolved:
refs.add(resolved)
modified = True
if modified:
script_value['recordRef'] = list(refs)
Expand All @@ -1453,8 +1560,7 @@ def execute(self, context: KeeperParams, **kwargs):
if not modified:
raise base.CommandError('Nothing to do')

record_management.update_record(vault, record)
vault.sync_data = True
_save_pam_typed_record(vault, record)


class PAMScriptDeleteCommand(base.ArgparseCommand):
Expand All @@ -1480,35 +1586,17 @@ def execute(self, context: KeeperParams, **kwargs):
if not script_name:
raise base.CommandError('"script" argument is required')

records = list(vault.vault_data.find_records(criteria=record_name, record_version=3, record_type=('pamUser', 'pamDirectory')))
if len(records) == 0:
raise base.CommandError(f'Record "{record_name}" not found')
if len(records) > 1:
raise base.CommandError(f'Record "{record_name}" is not unique. Use record UID.')
record = vault.vault_data.load_record(records[0].record_uid)
record = _get_unique_pam_script_record(vault, record_name)
if not isinstance(record, vault_record.TypedRecord):
raise base.CommandError(f'Record "{record.title}" is not a rotation record.')

script_field = next((x for x in record.fields if x.type == 'script'), None)
if script_field is None:
raise base.CommandError(f'Record "{record.title}" has no rotation scripts.')
script_value = next((x for x in script_field.value if x.get('fileRef') == script_name), None)
if script_value is None:
s_name = script_name.casefold()
for x in script_field.value:
file_uid = x.get('fileRef')
file_record = vault.vault_data.load_record(file_uid)
if isinstance(file_record, vault_record.FileRecord):
if file_record.record_uid == s_name:
script_value = x
break
elif file_record.title.casefold() == s_name:
script_value = x
break
script_value = _find_script_value(vault, script_field, script_name)

if not isinstance(script_value, dict):
raise base.CommandError(f'Record "{record.title}" does not have script "{script_name}"')

script_field.value.remove(script_value)
record_management.update_record(vault, record)
vault.sync_data = True
_save_pam_typed_record(vault, record)
Loading