From a4cd46adf79f638ed7b1bbec32583449c1efda2a Mon Sep 17 00:00:00 2001 From: sshrushanth-ks Date: Mon, 24 Aug 2026 10:44:24 +0530 Subject: [PATCH 1/5] fix validation for passwords and passphrase in record creation and updation --- unit-tests/test_nested_share_folder.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/unit-tests/test_nested_share_folder.py b/unit-tests/test_nested_share_folder.py index 5e5116a40..56bbb4dce 100644 --- a/unit-tests/test_nested_share_folder.py +++ b/unit-tests/test_nested_share_folder.py @@ -635,7 +635,6 @@ def test_update_record_allows_manual_weak_password(self, mock_perm, mock_update) cmd = NestedShareRecordUpdateCommand() cmd.execute(params, record_uids=[ruid], fields=['password=abc'], force=False) mock_update.assert_called_once() - @patch('keepercommander.commands.nested_share_folder.record_commands._nsf.update_record_v3') @patch('keepercommander.commands.nested_share_folder.helpers.check_record_edit_permission') def test_update_record_gen_password_uses_password_policy(self, mock_perm, mock_update): @@ -837,7 +836,6 @@ def test_update_rejects_missing_record_data(self, mock_perm, mock_update): cmd.execute(params, record_uids=[ruid], title='X', fields=[]) self.assertIn('not available', str(ctx.exception).lower()) mock_update.assert_not_called() - def test_update_rejects_invalid_field_spec(self): from keepercommander.commands.nested_share_folder import NestedShareRecordUpdateCommand cmd = NestedShareRecordUpdateCommand() From 5f696ccef780266bed2432788a979c18f0058cb5 Mon Sep 17 00:00:00 2001 From: sshrushanth-ks Date: Thu, 22 Oct 2026 12:05:20 +0530 Subject: [PATCH 2/5] Apply domain-specific policies to generated and manual passwords Select generator policies by record URL and validate generated passwords and passphrases with their respective rules. Apply explicit domain policies to manually entered passwords and passphrases, while leaving manual entries unchecked when no domain policy matches. Cover record add and update flows, including legacy and nested records. Add regression tests for policy selection and manual password validation. --- .../nested_share_folder/record_commands.py | 9 +- keepercommander/commands/record_edit.py | 82 +++++- keepercommander/commands/recordv2.py | 49 +++- keepercommander/commands/recordv3.py | 51 +++- keepercommander/enforcement.py | 206 ++++++++++++++- unit-tests/test_passphrase_enforcement.py | 236 +++++++++++++++++- 6 files changed, 603 insertions(+), 30 deletions(-) diff --git a/keepercommander/commands/nested_share_folder/record_commands.py b/keepercommander/commands/nested_share_folder/record_commands.py index 4748ac7f7..369e184bc 100644 --- a/keepercommander/commands/nested_share_folder/record_commands.py +++ b/keepercommander/commands/nested_share_folder/record_commands.py @@ -97,7 +97,7 @@ def execute(self, params, **kwargs): RecordTypeEnforcer.enforce(params, record_type, 'nsf-record-add') self.warnings.clear() - self._password_policy = PasswordComplexityEnforcer.get_policy(params) + self.reset_password_policy(params) notes = kwargs.get('notes') record_fields, add_attachments = _parse_field_specs(kwargs.get('fields', []), 'nsf-record-add') @@ -107,6 +107,7 @@ def execute(self, params, **kwargs): params, record_type, title, notes, record_fields) if self.abort_if_errors(): return + self.apply_manual_password_policy_warnings(params, data) if self.warnings: for w in self.warnings: logging.warning(w) @@ -250,7 +251,7 @@ def execute(self, params, **kwargs): raise CommandError('nsf-record-update', 'Record UID is required (use -r or --record)') self.warnings.clear() - self._password_policy = PasswordComplexityEnforcer.get_policy(params) + self.reset_password_policy(params) record_type = kwargs.get('record_type') rt_fields = None @@ -279,6 +280,8 @@ def execute(self, params, **kwargs): for identifier in record_uids: self.warnings.clear() self.errors.clear() + self._manual_password_values = [] + self._generated_password_values = [] record_uid = _nsf.resolve_nested_share_record_uid(params, identifier) if not record_uid: raise CommandError('nsf-record-update', @@ -295,6 +298,8 @@ def execute(self, params, **kwargs): if self.abort_if_errors(): continue + self.apply_manual_password_policy_warnings(params, record) + if self.warnings: for w in self.warnings: logging.warning(w) diff --git a/keepercommander/commands/record_edit.py b/keepercommander/commands/record_edit.py index e4f233105..43841b079 100644 --- a/keepercommander/commands/record_edit.py +++ b/keepercommander/commands/record_edit.py @@ -244,6 +244,9 @@ def __init__(self): self.warnings = [] self.errors = [] self._password_policy = None # type: Optional[Dict[str, Any]] + self._password_params = None + self._manual_password_values = [] + self._generated_password_values = [] def on_warning(self, message): if message: @@ -266,13 +269,62 @@ def abort_if_errors(self): def on_info(self, message): logging.info(message) + def reset_password_policy(self, params): + self._password_params = params + self._password_policy = PasswordComplexityEnforcer.get_policy(params) + self._manual_password_values = [] + self._generated_password_values = [] + + @staticmethod + def _url_override(fields): + for parsed_field in reversed(fields or []): + if parsed_field.type != 'url': + continue + value = parsed_field.value + if value.startswith('$JSON:'): + try: + value = json.loads(value[6:]) + except (json.JSONDecodeError, TypeError): + pass + return True, value + return False, None + + def _configure_password_policy(self, record, fields): + password_params = getattr(self, '_password_params', None) + if password_params is None: + return + has_url_override, url_override = self._url_override(fields) + kwargs = {'url_override': url_override} if has_url_override else {} + self._password_policy = PasswordComplexityEnforcer.get_generator_policy( + password_params, record, **kwargs) + + def _remember_manual_password(self, value): + values = value if isinstance(value, list) else [value] + generated_values = getattr(self, '_generated_password_values', []) + manual_values = getattr(self, '_manual_password_values', None) + if manual_values is None: + self._manual_password_values = manual_values = [] + for password in values: + if isinstance(password, str) and password and password not in generated_values: + manual_values.append(password) + + def apply_manual_password_policy_warnings(self, params, source): + failures = PasswordComplexityEnforcer.validate_manual_passwords( + params, source, getattr(self, '_manual_password_values', [])) + for failure in failures: + self.on_warning(failure) + def validate_generated_password(self, password, algorithm): """Validate a generated password with the correct algorithm flag.""" - if not password or not self._password_policy: + generated_values = getattr(self, '_generated_password_values', None) + if generated_values is None: + self._generated_password_values = generated_values = [] + generated_values.append(password) + password_policy = getattr(self, '_password_policy', None) + if not password or not password_policy: return - allow_passphrase = algorithm == 'passphrase' - failures = PasswordComplexityEnforcer.validate_password( - password, self._password_policy, allow_passphrase_fallback=allow_passphrase) + failures = PasswordComplexityEnforcer.validate_generated_password_with_policy( + password, password_policy, algorithm) for failure in failures: self.on_warning(failure) @@ -331,6 +383,9 @@ def assign_legacy_fields(self, record, fields): if not isinstance(fields, list): raise ValueError('Fields parameter: expected array of strings') + self._configure_password_policy(record, fields) + self._manual_password_values = [] + self._generated_password_values = [] action_params = [] for parsed_field in fields: if parsed_field.type == 'login': @@ -348,8 +403,10 @@ def assign_legacy_fields(self, record, fields): elif self.is_base64_value(parsed_field.value, action_params): if action_params: record.password = action_params[0] + self._remember_manual_password(action_params[0]) else: record.password = parsed_field.value + self._remember_manual_password(parsed_field.value) elif parsed_field.type == 'url': record.link = parsed_field.value elif parsed_field.type == 'oneTimeCode': @@ -619,6 +676,9 @@ def assign_typed_fields(self, record, fields): if not isinstance(fields, list): raise ValueError('Fields parameter: expected array of fields') + self._configure_password_policy(record, fields) + self._manual_password_values = [] + self._generated_password_values = [] parsed_fields = collections.deque(fields) while len(parsed_fields) > 0: parsed_field = parsed_fields.popleft() @@ -678,8 +738,10 @@ def assign_typed_fields(self, record, fields): if parsed_field.value: value = None + generated_value = False action_params = [] if self.is_generate_value(parsed_field.value, action_params): + generated_value = True if record_field.type == 'password': algorithm, _ = generator.resolve_gen_password_algorithm(action_params) value, gen_error = self.generate_password(action_params, policy=self._password_policy) @@ -694,7 +756,7 @@ def assign_typed_fields(self, record, fields): should_encrypt = 'enc' in action_params passphrase = None if should_encrypt: - passphrase, gen_error = self.generate_password() + passphrase, gen_error = self.generate_password(policy=self._password_policy) if gen_error: self.on_error(gen_error) continue @@ -766,6 +828,8 @@ def assign_typed_fields(self, record, fields): else: self.on_warning(f'Unsupported field type: {record_field.type}') if value is not None: + if record_field.type == 'password' and not generated_value: + self._remember_manual_password(value) if isinstance(value, list): record_field.value.clear() record_field.value.extend(value) @@ -943,7 +1007,7 @@ def execute(self, params, **kwargs): raise CommandError('record-add', 'Record type parameter is required.') RecordTypeEnforcer.enforce(params, record_type, 'record-add') - self._password_policy = PasswordComplexityEnforcer.get_policy(params) + self.reset_password_policy(params) fields = kwargs.get('fields', []) # Filter out empty strings that might be introduced by copy-paste or line continuation issues @@ -992,6 +1056,8 @@ def execute(self, params, **kwargs): if self.abort_if_errors(): return + self.apply_manual_password_policy_warnings(params, record) + record_uid = str(kwargs.get('record_uid', '')) if RecordV3.is_valid_ref_uid(record_uid): record.record_uid = record_uid @@ -1360,7 +1426,7 @@ def execute(self, params, **kwargs): else: record_fields.append(parsed_field) - self._password_policy = PasswordComplexityEnforcer.get_policy(params) + self.reset_password_policy(params) if isinstance(record, vault.PasswordRecord): raise CommandError('record-update', 'Legacy record type is not supported. Convert the record to login record type.') @@ -1381,6 +1447,8 @@ def execute(self, params, **kwargs): if self.abort_if_errors(): return + self.apply_manual_password_policy_warnings(params, record) + ignore_warnings = kwargs.get('force') is True if len(self.warnings) > 0: for warning in self.warnings: diff --git a/keepercommander/commands/recordv2.py b/keepercommander/commands/recordv2.py index e382b774e..551bd6301 100644 --- a/keepercommander/commands/recordv2.py +++ b/keepercommander/commands/recordv2.py @@ -19,6 +19,7 @@ from .base import suppress_exit, raise_parse_exception, Command from .. import api, generator, utils, crypto +from ..enforcement import PasswordComplexityEnforcer from ..error import CommandError from ..params import LAST_RECORD_UID from ..record import Record @@ -76,6 +77,25 @@ def custom_field_value(value): # type: (any) -> str return value +def _enforce_password_policy(params, source, password, generated, command, force=False): + if generated: + policy = PasswordComplexityEnforcer.get_generator_policy(params, source) + failures = PasswordComplexityEnforcer.validate_generated_password_with_policy( + password, policy, 'rand') + message = 'Password does not meet enterprise complexity policy.' + else: + failures = PasswordComplexityEnforcer.validate_manual_passwords( + params, source, [password] if password else []) + message = 'Password does not meet the domain-specific complexity policy.' + if failures: + for failure in failures: + logging.warning(failure) + if not force: + if command == 'add': + message += ' Pass --force to bypass this policy.' + raise CommandError(command, message) + + class RecordAddCommand(Command, RecordUtils): def get_parser(self): return add_parser @@ -90,7 +110,7 @@ def execute(self, params, **kwargs): generate = kwargs['generate'] if 'generate' in kwargs else None if generate: - password = generator.generate(16) + password = None force = kwargs['force'] if 'force' in kwargs else None if not force: @@ -98,13 +118,22 @@ def execute(self, params, **kwargs): title = input('...' + 'Title: '.rjust(16)) if not login: login = input('...' + 'Login: '.rjust(16)) - if not password: + if not password and not generate: password = input('...' + 'Password: '.rjust(16)) if not url: url = input('...' + 'Login URL: '.rjust(16)) if not title: raise CommandError('add', 'Invalid title. Expected non-empty string.') + password_policy_source = {'url': url or ''} + if generate: + policy = PasswordComplexityEnforcer.get_generator_policy( + params, password_policy_source) + password = (generator.KeeperPasswordGenerator.create_from_policy(policy).generate() + if policy else generator.generate(16)) + _enforce_password_policy( + params, password_policy_source, password, bool(generate), 'add', force=force) + custom = [] if custom_list: if type(custom_list) == str: @@ -266,13 +295,14 @@ def execute(self, params, **kwargs): record.password = kwargs['password'] password_changed = record.password != last_password changed = True - else: - if kwargs.get('generate'): - record.password = generator.generate(16) - changed = password_changed = True if kwargs.get('url') is not None: record.login_url = kwargs['url'] changed = True + if kwargs.get('password') is None and kwargs.get('generate'): + policy = PasswordComplexityEnforcer.get_generator_policy(params, record) + record.password = (generator.KeeperPasswordGenerator.create_from_policy(policy).generate() + if policy else generator.generate(16)) + changed = password_changed = True if kwargs.get('notes') is not None: notes = kwargs['notes'] # type: str if notes: @@ -327,6 +357,13 @@ def execute(self, params, **kwargs): if deleted: changed = True + if kwargs.get('password') is not None: + _enforce_password_policy( + params, record, kwargs.get('password'), False, 'edit') + elif kwargs.get('generate'): + _enforce_password_policy( + params, record, record.password, True, 'edit') + if changed: api.update_record(params, record) if password_changed: diff --git a/keepercommander/commands/recordv3.py b/keepercommander/commands/recordv3.py index b55d55a70..91956a417 100644 --- a/keepercommander/commands/recordv3.py +++ b/keepercommander/commands/recordv3.py @@ -131,12 +131,17 @@ def register_command_info(aliases, command_info): # command_group.add_argument('-r', '--remove-type', dest='remove_type', action='store_true', help='delete custom record type') -def get_password_from_rules(generate_rules, generate_length): +def get_password_from_rules(generate_rules, generate_length, policy=None): if generate_rules: kpg = generator.KeeperPasswordGenerator.create_from_rules(generate_rules, length=generate_length) if kpg is None: logging.warning('Using default password complexity rules') - kpg = generator.KeeperPasswordGenerator(generate_length or DEFAULT_GENERATE_PASSWORD_LENGTH) + kpg = (generator.KeeperPasswordGenerator.create_from_policy( + policy, length_override=generate_length) if policy else + generator.KeeperPasswordGenerator(generate_length or DEFAULT_GENERATE_PASSWORD_LENGTH)) + elif policy: + kpg = generator.KeeperPasswordGenerator.create_from_policy( + policy, length_override=generate_length) else: kpg = generator.KeeperPasswordGenerator(generate_length or DEFAULT_GENERATE_PASSWORD_LENGTH) return kpg.generate() @@ -146,8 +151,11 @@ def enforce_generated_password_policy(params, data, command, generated, manual_p # Skip if user explicitly provided a password (overrides generation) if not generated or manual_password: return - pw_failures = PasswordComplexityEnforcer.validate_record( - params, data, allow_passphrase_fallback=False) + policy = PasswordComplexityEnforcer.get_generator_policy(params, data) + pw_failures = [] + for password in PasswordComplexityEnforcer.get_password_values(data): + pw_failures.extend(PasswordComplexityEnforcer.validate_generated_password_with_policy( + password, policy, 'rand')) if pw_failures: for failure in pw_failures: logging.warning(bcolors.WARNING + failure + bcolors.ENDC) @@ -158,6 +166,18 @@ def enforce_generated_password_policy(params, data, command, generated, manual_p 'Pass --force to bypass these warnings.') +def enforce_manual_password_policy(params, data, command, passwords, force=False): + failures = PasswordComplexityEnforcer.validate_manual_passwords(params, data, passwords) + if failures: + for failure in failures: + logging.warning(bcolors.WARNING + failure + bcolors.ENDC) + if not force: + raise CommandError( + command, + 'Password does not meet the domain-specific complexity policy. ' + 'Pass --force to bypass this policy.') + + class RecordAddCommand(Command, recordv2.RecordUtils): def get_parser(self): return add_parser @@ -423,12 +443,18 @@ def GCM_TAG_LEN(): return 16 manual_password = kwargs.get('password') password = manual_password if not password and (kwargs.get('generate') or kwargs.get('generate_rules') or kwargs.get('generate_length')): - password = get_password_from_rules(kwargs.get('generate_rules'), kwargs.get('generate_length')) + policy = PasswordComplexityEnforcer.get_generator_policy(params, data) + password = get_password_from_rules( + kwargs.get('generate_rules'), kwargs.get('generate_length'), policy) if password: data = recordv3.RecordV3.update_password(password, data, recordv3.RecordV3.get_record_type_definition(params, data)) generated = bool(kwargs.get('generate') or kwargs.get('generate_rules') or kwargs.get('generate_length')) enforce_generated_password_policy(params, data, 'add', generated, manual_password=manual_password, force=kwargs.get('force')) + manual_passwords = ([manual_password] if manual_password else + PasswordComplexityEnforcer.get_password_values(data) if not generated else []) + enforce_manual_password_policy( + params, data, 'add', manual_passwords, force=kwargs.get('force')) record_uid = api.generate_record_uid() logging.debug('Generated Record UID: %s', record_uid) @@ -666,13 +692,26 @@ def execute(self, params, **kwargs): manual_password = kwargs.get('password') password = manual_password if not password and generate: - password = get_password_from_rules(kwargs.get('generate_rules'), kwargs.get('generate_length')) + policy = PasswordComplexityEnforcer.get_generator_policy(params, data) + password = get_password_from_rules( + kwargs.get('generate_rules'), kwargs.get('generate_length'), policy) if password: record.password = password data = recordv3.RecordV3.update_password(password, data, recordv3.RecordV3.get_record_type_definition(params, data)) generated = bool(kwargs.get('generate') or kwargs.get('generate_rules') or kwargs.get('generate_length')) enforce_generated_password_policy(params, data, 'edit', generated, manual_password=manual_password, force=kwargs.get('force')) + if manual_password: + manual_passwords = [manual_password] + elif not generated: + manual_passwords = PasswordComplexityEnforcer.get_password_values(data) + for existing_password in PasswordComplexityEnforcer.get_password_values(record_data): + if existing_password in manual_passwords: + manual_passwords.remove(existing_password) + else: + manual_passwords = [] + enforce_manual_password_policy( + params, data, 'edit', manual_passwords, force=kwargs.get('force')) data_dict = json.loads(data) changed = rdata_dict != data_dict diff --git a/keepercommander/enforcement.py b/keepercommander/enforcement.py index a53e824eb..85ff1a147 100644 --- a/keepercommander/enforcement.py +++ b/keepercommander/enforcement.py @@ -17,6 +17,7 @@ import threading from datetime import datetime, timedelta from typing import Tuple, Optional, List, Dict, Any, Set +from urllib.parse import urlsplit from . import api, utils, crypto from .proto import APIRequest_pb2, record_pb2 @@ -67,6 +68,9 @@ def _coerce_int(value): return None +_DOMAIN_URL_UNSET = object() + + class MasterPasswordReentryEnforcer: """Handler for MASTER_PASSWORD_REENTRY enforcement with multi-factor authentication support.""" @@ -412,23 +416,211 @@ class PasswordComplexityEnforcer: ) @classmethod - def get_policy(cls, params): # type: (KeeperParams) -> Optional[Dict[str, Any]] + def get_policies(cls, params): # type: (KeeperParams) -> List[Dict[str, Any]] if not params or not params.enforcements: - return None + return [] raw = _find_enforcement_value(params.enforcements, cls._POLICY_KEY) if raw is None: - return None + return [] try: rules = json.loads(raw) if isinstance(raw, str) else raw except (json.JSONDecodeError, TypeError): logging.debug('Failed to parse %s enforcement', cls._POLICY_KEY) - return None - if isinstance(rules, list) and rules and isinstance(rules[0], dict): - return rules[0] + return [] + if isinstance(rules, list): + return [x for x in rules if isinstance(x, dict)] if isinstance(rules, dict): - return rules + return [rules] + return [] + + @classmethod + def _default_policy(cls, policies): # type: (List[Dict[str, Any]]) -> Optional[Dict[str, Any]] + if not policies: + return None + for policy in policies: + domains = policy.get('domains') + if isinstance(domains, list) and any( + isinstance(x, str) and x.strip().lower() == '_default_' for x in domains): + return policy + for policy in policies: + if not isinstance(policy.get('domains'), list): + return policy return None + @classmethod + def get_policy(cls, params): # type: (KeeperParams) -> Optional[Dict[str, Any]] + """Return the default password-generator policy.""" + return cls._default_policy(cls.get_policies(params)) + + @classmethod + def _normalize_domain(cls, value): # type: (Any) -> Optional[str] + if not isinstance(value, str): + return None + value = value.strip().lower().rstrip('.') + if not value or value == '_default_': + return None + if value.startswith('*.'): + value = value[2:] + try: + candidate = value if '://' in value else '//' + value + host = urlsplit(candidate).hostname + except ValueError: + return None + if not host: + return None + try: + return host.encode('idna').decode('ascii').lower().rstrip('.') + except (UnicodeError, ValueError): + return None + + @classmethod + def _url_values(cls, source): # type: (Any) -> List[Any] + urls = [] + if source is None: + return urls + + for attr in ('link', 'login_url', 'url'): + value = getattr(source, attr, None) + if value: + urls.append(value) + + if hasattr(source, 'fields') or hasattr(source, 'custom'): + for field in itertools.chain( + getattr(source, 'fields', None) or [], + getattr(source, 'custom', None) or []): + if getattr(field, 'type', None) == 'url': + value = getattr(field, 'value', None) + urls.extend(value if isinstance(value, list) else [value]) + return urls + + data = source + if isinstance(data, str): + try: + data = json.loads(data) + except (json.JSONDecodeError, TypeError): + return urls + if isinstance(data, dict): + for key in ('url', 'link', 'login_url'): + if data.get(key): + urls.append(data[key]) + for field in itertools.chain(data.get('fields') or [], data.get('custom') or []): + if isinstance(field, dict) and field.get('type') == 'url': + value = field.get('value') + urls.extend(value if isinstance(value, list) else [value]) + return urls + + @classmethod + def _source_domains(cls, source, url_override=_DOMAIN_URL_UNSET): # type: (Any, Any) -> Set[str] + urls = cls._url_values(source) + if url_override is not _DOMAIN_URL_UNSET: + urls = [url_override] if url_override else [] + domains = set() + for value in urls: + if isinstance(value, dict): + value = value.get('url') or value.get('hostName') + domain = cls._normalize_domain(value) + if domain: + domains.add(domain) + return domains + + @classmethod + def _domain_policy_for_source(cls, policies, source, url_override=_DOMAIN_URL_UNSET): + # type: (List[Dict[str, Any]], Any, Any) -> Optional[Dict[str, Any]] + source_domains = cls._source_domains(source, url_override) + if not source_domains: + return None + + matches = [] + for policy in policies or []: + domains = policy.get('domains') + if not isinstance(domains, list): + continue + normalized_domains = [ + cls._normalize_domain(domain) + for domain in domains + if isinstance(domain, str) and domain.strip().lower() != '_default_' + ] + normalized_domains = [x for x in normalized_domains if x] + matched_domains = [ + configured for configured in normalized_domains + if any(host == configured or host.endswith('.' + configured) + for host in source_domains) + ] + if matched_domains: + matches.append((max(len(x) for x in matched_domains), policy)) + if not matches: + return None + return max(matches, key=lambda x: x[0])[1] + + @classmethod + def get_generator_policy(cls, params, source=None, url_override=_DOMAIN_URL_UNSET): + # type: (KeeperParams, Any, Any) -> Optional[Dict[str, Any]] + policies = cls.get_policies(params) + default_policy = cls._default_policy(policies) + domain_policy = cls._domain_policy_for_source(policies, source, url_override) + if not domain_policy: + return default_policy + merged = dict(default_policy or {}) + merged.update(domain_policy) + return merged + + @classmethod + def get_domain_specific_policy(cls, params, source, url_override=_DOMAIN_URL_UNSET): + # type: (KeeperParams, Any, Any) -> Optional[Dict[str, Any]] + """Return a matching explicit domain policy, excluding the generator default.""" + return cls._domain_policy_for_source( + cls.get_policies(params), source, url_override) + + @classmethod + def get_password_values(cls, source): # type: (Any) -> List[str] + return cls._extract_passwords(source) + + @classmethod + def validate_generated_password_with_policy(cls, password, policy, algorithm): + # type: (str, Optional[Dict[str, Any]], str) -> List[str] + if not policy or not password: + return [] + if algorithm == 'passphrase' and policy.get('passphrase-allow') is not False: + return cls.validate_passphrase(password, policy) + return cls.validate_password(password, policy, allow_passphrase_fallback=False) + + @classmethod + def validate_manual_passwords(cls, params, source, passwords=None): + # type: (KeeperParams, Any, Optional[List[str]]) -> List[str] + """Validate explicitly entered password values against matching domain rules only.""" + policy = cls.get_domain_specific_policy(params, source) + if not policy: + return [] + values = passwords if passwords is not None else cls._extract_passwords(source) + failures = [] + passphrase_policy_present = any( + key.startswith('passphrase-') for key in policy) + for password in values: + is_passphrase = ( + passphrase_policy_present + and policy.get('passphrase-allow') is not False + and cls._looks_like_passphrase(password, policy) + ) + if is_passphrase: + failures.extend(cls.validate_passphrase(password, policy)) + else: + failures.extend(cls.validate_password( + password, policy, allow_passphrase_fallback=False)) + return failures + + @staticmethod + def _looks_like_passphrase(password, policy): # type: (Any, Dict[str, Any]) -> bool + if not isinstance(password, str) or not password: + return False + separators = PasswordComplexityEnforcer._normalize_passphrase_separators(policy) + if not separators: + return False + separator = next((char for char in separators if char in password), None) + if separator is None: + return False + words = password.split(separator) + return bool(words) and all(re.fullmatch(r'[A-Za-z]{3,}[0-9]?', word) for word in words) + @classmethod def _normalize_passphrase_separators(cls, policy): # type: (Dict[str, Any]) -> Optional[str] raw = policy.get('passphrase-separator') diff --git a/unit-tests/test_passphrase_enforcement.py b/unit-tests/test_passphrase_enforcement.py index 9bc0f6471..403670d61 100644 --- a/unit-tests/test_passphrase_enforcement.py +++ b/unit-tests/test_passphrase_enforcement.py @@ -1,10 +1,15 @@ +import json from types import SimpleNamespace from unittest import TestCase -from keepercommander.commands.record_edit import RecordEditMixin -from keepercommander.commands.recordv3 import enforce_generated_password_policy +from keepercommander.commands.record_edit import ParsedFieldValue, RecordEditMixin +from keepercommander.commands.recordv3 import ( + enforce_generated_password_policy, + enforce_manual_password_policy, +) from keepercommander.enforcement import PasswordComplexityEnforcer from keepercommander.error import CommandError +from keepercommander import vault STRICT_RANDOM_POLICY = { @@ -142,3 +147,230 @@ def test_explicit_password_overrides_generation_for_policy(self): enforce_generated_password_policy( self._params(), self._record(), 'edit', generated=True, manual_password='explicit', force=False) + + +class TestDomainPasswordPolicy(TestCase): + + @staticmethod + def _params(policies): + return SimpleNamespace(enforcements={ + 'jsons': [{ + 'key': 'generated_password_complexity', + 'value': json.dumps(policies), + }], + }) + + @staticmethod + def _record(url=None, password='weak'): + return { + 'type': 'login', + 'fields': ([{'type': 'url', 'value': [url]}] if url else []) + [ + {'type': 'password', 'value': [password]}], + } + + def test_generator_policy_merges_matching_domain_over_default(self): + params = self._params([ + { + 'domains': ['_default_'], + 'length': 12, + 'lower-use': True, + 'lower-min': 2, + }, + { + 'domains': ['example.com'], + 'length': 18, + 'upper-use': True, + 'upper-min': 2, + }, + ]) + + policy = PasswordComplexityEnforcer.get_generator_policy( + params, self._record('https://login.example.com/account')) + + self.assertEqual(policy['length'], 18) + self.assertEqual(policy['lower-min'], 2) + self.assertEqual(policy['upper-min'], 2) + + def test_manual_password_ignores_generator_default_policy(self): + params = self._params([{ + 'domains': ['_default_'], + 'length': 40, + 'upper-use': True, + 'upper-min': 4, + 'passphrase-allow': True, + 'passphrase-length': 6, + }]) + + failures = PasswordComplexityEnforcer.validate_manual_passwords( + params, self._record('https://example.com')) + + self.assertEqual(failures, []) + + def test_manual_password_uses_matching_domain_policy(self): + params = self._params([ + { + 'domains': ['_default_'], + 'length': 40, + }, + { + 'domains': ['google.com'], + 'length': 12, + 'upper-use': True, + 'upper-min': 1, + 'digit-use': True, + 'digit-min': 1, + 'special-use': True, + 'special-min': 1, + 'passphrase-allow': True, + 'passphrase-length': 5, + 'passphrase-capitalize': False, + 'passphrase-number': False, + 'passphrase-separator': '-', + }, + ]) + + failures = PasswordComplexityEnforcer.validate_manual_passwords( + params, self._record('google.com')) + + self.assertTrue(any('at least 12 characters' in failure for failure in failures)) + self.assertTrue(any('uppercase character' in failure for failure in failures)) + self.assertTrue(any('digit character' in failure for failure in failures)) + self.assertTrue(any('special character' in failure for failure in failures)) + self.assertFalse(any('Passphrase' in failure for failure in failures)) + + def test_manual_passphrase_uses_matching_domain_passphrase_rules(self): + params = self._params([{ + 'domains': ['example.com'], + 'length': 80, + 'upper-use': True, + 'upper-min': 20, + 'passphrase-allow': True, + 'passphrase-length': 3, + 'passphrase-capitalize': False, + 'passphrase-number': False, + 'passphrase-separator': '-', + }]) + + failures = PasswordComplexityEnforcer.validate_manual_passwords( + params, + self._record('https://example.com', 'alpha-bravo-charlie')) + + self.assertEqual(failures, []) + + def test_manual_password_with_dollar_and_digit_is_not_mistaken_for_passphrase(self): + params = self._params([{ + 'domains': ['google.com'], + 'length': 20, + 'upper-use': True, + 'upper-min': 1, + 'digit-use': True, + 'digit-min': 1, + 'special-use': True, + 'special-min': 1, + 'passphrase-allow': True, + 'passphrase-length': 5, + 'passphrase-capitalize': False, + 'passphrase-number': False, + 'passphrase-separator': '-', + }]) + + failures = PasswordComplexityEnforcer.validate_manual_passwords( + params, + self._record('google.com', 'asd$2Egjrstkdsbtvoastasfawf')) + + self.assertEqual(failures, []) + + def test_manual_password_ignores_unmatched_domain_policy(self): + params = self._params([{ + 'domains': ['other.example'], + 'length': 40, + }]) + + failures = PasswordComplexityEnforcer.validate_manual_passwords( + params, self._record('https://example.com')) + + self.assertEqual(failures, []) + + def test_typed_manual_password_uses_url_supplied_in_same_update(self): + params = self._params([{ + 'domains': ['example.com'], + 'length': 12, + 'passphrase-allow': False, + }]) + mixin = RecordEditMixin() + mixin.reset_password_policy(params) + record = vault.TypedRecord() + record.type_name = 'login' + record.fields.extend([ + vault.TypedField.new_field('url', None, ''), + vault.TypedField.new_field('password', None, ''), + ]) + + # The password arrives before the URL; policy selection must still use + # the URL being written by this update. + mixin.assign_typed_fields(record, [ + ParsedFieldValue('', 'password', '', 'weak'), + ParsedFieldValue('', 'url', '', 'https://example.com/login'), + ]) + mixin.apply_manual_password_policy_warnings(params, record) + + self.assertTrue(any('at least 12 characters' in warning for warning in mixin.warnings)) + + def test_generated_passphrase_uses_only_passphrase_rules(self): + policy = { + 'length': 80, + 'upper-use': True, + 'upper-min': 20, + 'passphrase-allow': True, + 'passphrase-length': 5, + 'passphrase-capitalize': False, + 'passphrase-number': False, + 'passphrase-separator': '-', + } + passphrase = 'alpha-bravo-charlie-delta-echo' + + failures = PasswordComplexityEnforcer.validate_generated_password_with_policy( + passphrase, policy, 'passphrase') + + self.assertEqual(failures, []) + + def test_generated_password_does_not_fall_back_to_passphrase_rules(self): + policy = { + 'length': 80, + 'passphrase-allow': True, + 'passphrase-length': 5, + 'passphrase-separator': '-', + } + + failures = PasswordComplexityEnforcer.validate_generated_password_with_policy( + 'alpha-bravo-charlie-delta-echo', policy, 'rand') + + self.assertTrue(any('at least 80 characters' in failure for failure in failures)) + + def test_recordv3_generated_password_uses_matching_domain_policy(self): + params = self._params([ + {'domains': ['_default_'], 'length': 4}, + {'domains': ['example.com'], 'length': 20}, + ]) + + with self.assertRaises(CommandError): + enforce_generated_password_policy( + params, self._record('https://login.example.com'), 'add', + generated=True) + + def test_recordv3_manual_password_requires_a_matching_domain_policy(self): + default_only = self._params([{ + 'domains': ['_default_'], + 'length': 20, + }]) + enforce_manual_password_policy( + default_only, self._record('https://example.com'), 'add', ['weak']) + + domain_specific = self._params([{ + 'domains': ['example.com'], + 'length': 20, + 'passphrase-allow': False, + }]) + with self.assertRaises(CommandError): + enforce_manual_password_policy( + domain_specific, self._record('https://example.com'), 'add', ['weak']) From 6e009a7f807f3c0ecfe8d223c68e80f215b4c8f8 Mon Sep 17 00:00:00 2001 From: sshrushanth-ks Date: Thu, 22 Oct 2026 12:31:31 +0530 Subject: [PATCH 3/5] Handle Unicode passphrases in domain policy validation Normalize passphrase words to NFC and accept Unicode letters during manual passphrase validation. Remove the unused NSF field resolver and add regression coverage for domain-policy enforcement in NSF updates. --- .../nested_share_folder/record_commands.py | 30 +---------- keepercommander/enforcement.py | 51 ++++++++++++------- unit-tests/test_nested_share_folder.py | 32 ++++++++++++ unit-tests/test_passphrase_enforcement.py | 20 ++++++++ 4 files changed, 85 insertions(+), 48 deletions(-) diff --git a/keepercommander/commands/nested_share_folder/record_commands.py b/keepercommander/commands/nested_share_folder/record_commands.py index 369e184bc..6ac5ebd6c 100644 --- a/keepercommander/commands/nested_share_folder/record_commands.py +++ b/keepercommander/commands/nested_share_folder/record_commands.py @@ -25,7 +25,7 @@ from ..record_edit import RecordEditMixin, record_fields_description from ...enforcement import PasswordComplexityEnforcer, RecordTypeEnforcer from ...error import CommandError -from ... import generator, nested_share_folder as _nsf, vault, vault_extensions +from ... import nested_share_folder as _nsf, vault, vault_extensions from .helpers import ( resolve_folder_uid, command_error_handler, check_result, check_record_edit_permission, check_record_delete_permission, @@ -213,34 +213,6 @@ def __init__(self): def get_parser(self): return nested_share_record_update_parser - def _resolve_field_value(self, parsed): - raw = parsed.value - if not raw: - return raw - - action_params = [] - if self.is_json_value(raw, action_params): - return action_params[0] if action_params else None - action_params.clear() - if self.is_generate_value(raw, action_params): - if self.warn_wrong_password_gen_field(parsed): - return None - if parsed.type == 'password': - algorithm, _ = generator.resolve_gen_password_algorithm(action_params) - password, gen_error = self.generate_password(action_params, policy=self._password_policy) - if gen_error: - self.on_error(gen_error) - return None - if password is not None: - self.validate_generated_password(password, algorithm) - return password - if parsed.type in ('oneTimeCode', 'otp'): - return self.generate_totp_url() - return raw - action_params.clear() - if self.is_base64_value(raw, action_params): - return action_params[0] if action_params else None - return raw def execute(self, params, **kwargs): if kwargs.get('syntax_help'): print(record_fields_description) diff --git a/keepercommander/enforcement.py b/keepercommander/enforcement.py index 85ff1a147..5a9adb6db 100644 --- a/keepercommander/enforcement.py +++ b/keepercommander/enforcement.py @@ -15,6 +15,7 @@ import getpass import re import threading +import unicodedata from datetime import datetime, timedelta from typing import Tuple, Optional, List, Dict, Any, Set from urllib.parse import urlsplit @@ -608,18 +609,27 @@ def validate_manual_passwords(cls, params, source, passwords=None): password, policy, allow_passphrase_fallback=False)) return failures - @staticmethod - def _looks_like_passphrase(password, policy): # type: (Any, Dict[str, Any]) -> bool + @classmethod + def _looks_like_passphrase(cls, password, policy): # type: (Any, Dict[str, Any]) -> bool if not isinstance(password, str) or not password: return False - separators = PasswordComplexityEnforcer._normalize_passphrase_separators(policy) + separators = cls._normalize_passphrase_separators(policy) if not separators: return False separator = next((char for char in separators if char in password), None) if separator is None: return False words = password.split(separator) - return bool(words) and all(re.fullmatch(r'[A-Za-z]{3,}[0-9]?', word) for word in words) + return bool(words) and all(cls._passphrase_word_letters(word, allow_trailing_number=True) + for word in words) + + @staticmethod + def _passphrase_word_letters(word, allow_trailing_number=False): + # type: (str, bool) -> Optional[str] + word = unicodedata.normalize('NFC', word) + if allow_trailing_number and word and word[-1] in '0123456789': + word = word[:-1] + return word if len(word) >= 3 and word.isalpha() else None @classmethod def _normalize_passphrase_separators(cls, policy): # type: (Dict[str, Any]) -> Optional[str] @@ -669,22 +679,25 @@ def validate_passphrase(cls, password, policy): # type: (str, Dict[str, Any]) capitalize = bool(policy.get('passphrase-capitalize', DEFAULT_PASSPHRASE_CAPITALIZE)) append_number = bool(policy.get('passphrase-number', DEFAULT_PASSPHRASE_NUMBER)) - if capitalize: - word_re = re.compile(r'^[A-Z][a-z]{2,}$') - if append_number: - first_re = re.compile(r'^[A-Z][a-z]{2,}[0-9]$') - else: - first_re = re.compile(r'^[A-Z][a-z]{2,}[0-9]?$') - else: - word_re = re.compile(r'^[A-Za-z]{3,}$') - if append_number: - first_re = re.compile(r'^[A-Za-z]{3,}[0-9]$') - else: - first_re = re.compile(r'^[A-Za-z]{3,}[0-9]?$') - for index, word in enumerate(words): - pattern = first_re if index == 0 else word_re - if not pattern.match(word): + normalized_word = unicodedata.normalize('NFC', word) + is_first = index == 0 + has_number = bool(normalized_word and normalized_word[-1] in '0123456789') + if is_first and append_number: + if not has_number: + failures.append('First passphrase word must end with a digit (0-9).') + break + normalized_word = normalized_word[:-1] + elif is_first and has_number: + normalized_word = normalized_word[:-1] + + letters = cls._passphrase_word_letters(normalized_word) + correctly_capitalized = ( + not capitalize or + (bool(letters) and letters[0].isupper() + and all(char.islower() for char in letters[1:])) + ) + if not letters or not correctly_capitalized: if index == 0 and append_number: failures.append('First passphrase word must end with a digit (0-9).') elif capitalize: diff --git a/unit-tests/test_nested_share_folder.py b/unit-tests/test_nested_share_folder.py index 56bbb4dce..59071e97a 100644 --- a/unit-tests/test_nested_share_folder.py +++ b/unit-tests/test_nested_share_folder.py @@ -635,6 +635,38 @@ def test_update_record_allows_manual_weak_password(self, mock_perm, mock_update) cmd = NestedShareRecordUpdateCommand() cmd.execute(params, record_uids=[ruid], fields=['password=abc'], force=False) mock_update.assert_called_once() + + @patch('keepercommander.commands.nested_share_folder.record_commands._nsf.update_record_v3') + @patch('keepercommander.commands.nested_share_folder.helpers.check_record_edit_permission') + def test_update_record_enforces_matching_domain_policy_for_manual_password( + self, mock_perm, mock_update): + from keepercommander.commands.nested_share_folder import NestedShareRecordUpdateCommand + ruid, robj = _make_record() + params = _make_params( + nested_share_records={ruid: robj}, + record_cache={ruid: {'revision': 1, 'data_unencrypted': json.dumps({ + 'type': 'login', 'title': 'Old', + 'fields': [{'type': 'password', 'value': ['ExistingPass123']}], + })}}, + enforcements={ + 'jsons': [{ + 'key': 'generated_password_complexity', + 'value': json.dumps([{ + 'domains': ['example.com'], + 'length': 12, + 'passphrase-allow': False, + }]), + }], + }, + ) + + NestedShareRecordUpdateCommand().execute( + params, record_uids=[ruid], + fields=['password=weak', 'url=https://example.com/login'], + force=False) + + mock_update.assert_not_called() + @patch('keepercommander.commands.nested_share_folder.record_commands._nsf.update_record_v3') @patch('keepercommander.commands.nested_share_folder.helpers.check_record_edit_permission') def test_update_record_gen_password_uses_password_policy(self, mock_perm, mock_update): diff --git a/unit-tests/test_passphrase_enforcement.py b/unit-tests/test_passphrase_enforcement.py index 403670d61..3d75b1c6a 100644 --- a/unit-tests/test_passphrase_enforcement.py +++ b/unit-tests/test_passphrase_enforcement.py @@ -257,6 +257,26 @@ def test_manual_passphrase_uses_matching_domain_passphrase_rules(self): self.assertEqual(failures, []) + def test_manual_passphrase_accepts_unicode_words(self): + for passphrase, capitalize in ( + ('élan-café-über', False), + ('Élan-Café-Über', True)): + with self.subTest(passphrase=passphrase): + params = self._params([{ + 'domains': ['example.com'], + 'passphrase-allow': True, + 'passphrase-length': 3, + 'passphrase-capitalize': capitalize, + 'passphrase-number': False, + 'passphrase-separator': '-', + }]) + + failures = PasswordComplexityEnforcer.validate_manual_passwords( + params, + self._record('https://example.com', passphrase)) + + self.assertEqual(failures, []) + def test_manual_password_with_dollar_and_digit_is_not_mistaken_for_passphrase(self): params = self._params([{ 'domains': ['google.com'], From 20a21c3e2ca2304fb3c257a8f6c727d3650b219a Mon Sep 17 00:00:00 2001 From: sshrushanth-ks Date: Fri, 23 Oct 2026 13:35:47 +0530 Subject: [PATCH 4/5] Apply domain password rules to manual entries without inferring passphrase intent from separators. Keep passphrase rules for generated passphrases and add regression coverage for hyphenated passwords. --- keepercommander/enforcement.py | 34 +++-------------------- unit-tests/test_passphrase_enforcement.py | 31 +++++++++++++++++---- 2 files changed, 29 insertions(+), 36 deletions(-) diff --git a/keepercommander/enforcement.py b/keepercommander/enforcement.py index 5a9adb6db..86f624842 100644 --- a/keepercommander/enforcement.py +++ b/keepercommander/enforcement.py @@ -594,41 +594,15 @@ def validate_manual_passwords(cls, params, source, passwords=None): return [] values = passwords if passwords is not None else cls._extract_passwords(source) failures = [] - passphrase_policy_present = any( - key.startswith('passphrase-') for key in policy) for password in values: - is_passphrase = ( - passphrase_policy_present - and policy.get('passphrase-allow') is not False - and cls._looks_like_passphrase(password, policy) - ) - if is_passphrase: - failures.extend(cls.validate_passphrase(password, policy)) - else: - failures.extend(cls.validate_password( - password, policy, allow_passphrase_fallback=False)) + failures.extend(cls.validate_password( + password, policy, allow_passphrase_fallback=False)) return failures - @classmethod - def _looks_like_passphrase(cls, password, policy): # type: (Any, Dict[str, Any]) -> bool - if not isinstance(password, str) or not password: - return False - separators = cls._normalize_passphrase_separators(policy) - if not separators: - return False - separator = next((char for char in separators if char in password), None) - if separator is None: - return False - words = password.split(separator) - return bool(words) and all(cls._passphrase_word_letters(word, allow_trailing_number=True) - for word in words) - @staticmethod - def _passphrase_word_letters(word, allow_trailing_number=False): - # type: (str, bool) -> Optional[str] + def _passphrase_word_letters(word): + # type: (str) -> Optional[str] word = unicodedata.normalize('NFC', word) - if allow_trailing_number and word and word[-1] in '0123456789': - word = word[:-1] return word if len(word) >= 3 and word.isalpha() else None @classmethod diff --git a/unit-tests/test_passphrase_enforcement.py b/unit-tests/test_passphrase_enforcement.py index 3d75b1c6a..c23b6dc07 100644 --- a/unit-tests/test_passphrase_enforcement.py +++ b/unit-tests/test_passphrase_enforcement.py @@ -238,7 +238,7 @@ def test_manual_password_uses_matching_domain_policy(self): self.assertTrue(any('special character' in failure for failure in failures)) self.assertFalse(any('Passphrase' in failure for failure in failures)) - def test_manual_passphrase_uses_matching_domain_passphrase_rules(self): + def test_manual_passphrase_uses_matching_domain_password_rules(self): params = self._params([{ 'domains': ['example.com'], 'length': 80, @@ -255,9 +255,11 @@ def test_manual_passphrase_uses_matching_domain_passphrase_rules(self): params, self._record('https://example.com', 'alpha-bravo-charlie')) - self.assertEqual(failures, []) + self.assertTrue(any('at least 80 characters' in failure for failure in failures)) + self.assertTrue(any('uppercase character' in failure for failure in failures)) + self.assertFalse(any('Passphrase' in failure for failure in failures)) - def test_manual_passphrase_accepts_unicode_words(self): + def test_passphrase_validator_accepts_unicode_words(self): for passphrase, capitalize in ( ('élan-café-über', False), ('Élan-Café-Über', True)): @@ -271,12 +273,29 @@ def test_manual_passphrase_accepts_unicode_words(self): 'passphrase-separator': '-', }]) - failures = PasswordComplexityEnforcer.validate_manual_passwords( - params, - self._record('https://example.com', passphrase)) + policy = json.loads(params.enforcements['jsons'][0]['value'])[0] + failures = PasswordComplexityEnforcer.validate_passphrase( + passphrase, policy) self.assertEqual(failures, []) + def test_manual_hyphenated_password_uses_password_rules_not_passphrase_rules(self): + params = self._params([{ + 'domains': ['example.com'], + 'length': 20, + 'passphrase-allow': True, + 'passphrase-length': 5, + 'passphrase-capitalize': False, + 'passphrase-number': False, + 'passphrase-separator': '-', + }]) + + failures = PasswordComplexityEnforcer.validate_manual_passwords( + params, + self._record('https://example.com', 'mysecret-horse-battery')) + + self.assertEqual(failures, []) + def test_manual_password_with_dollar_and_digit_is_not_mistaken_for_passphrase(self): params = self._params([{ 'domains': ['google.com'], From 6274a4b31ed21974c28eb800c988b8bffd681d78 Mon Sep 17 00:00:00 2001 From: sshrushanth-ks Date: Fri, 30 Oct 2026 17:41:13 +0530 Subject: [PATCH 5/5] Addressed review comments: MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Use matching domain settings for passphrase generation and allow manual entries that satisfy the domain’s password or permitted passphrase rules. Prefer the primary record URL, add a policy override for legacy edits, and revalidate manual passwords on NSF update retries. Add regression coverage for policy selection, URL precedence, overrides, and retry validation. --- .../nested_share_folder/record_commands.py | 1 + keepercommander/commands/record_edit.py | 27 ++- keepercommander/commands/recordv2.py | 16 +- keepercommander/enforcement.py | 35 +++- keepercommander/generator.py | 6 +- unit-tests/test_nested_share_folder.py | 51 ++++++ unit-tests/test_passphrase_enforcement.py | 159 +++++++++++++++++- unit-tests/test_passphrase_generator.py | 8 +- 8 files changed, 270 insertions(+), 33 deletions(-) diff --git a/keepercommander/commands/nested_share_folder/record_commands.py b/keepercommander/commands/nested_share_folder/record_commands.py index 6ac5ebd6c..1fc8c64c6 100644 --- a/keepercommander/commands/nested_share_folder/record_commands.py +++ b/keepercommander/commands/nested_share_folder/record_commands.py @@ -292,6 +292,7 @@ def execute(self, params, **kwargs): record_type, rt_fields, record_fields) if self.abort_if_errors(): continue + self.apply_manual_password_policy_warnings(params, record) if self.warnings: for w in self.warnings: logging.warning(w) diff --git a/keepercommander/commands/record_edit.py b/keepercommander/commands/record_edit.py index 43841b079..84b71aac6 100644 --- a/keepercommander/commands/record_edit.py +++ b/keepercommander/commands/record_edit.py @@ -276,12 +276,25 @@ def reset_password_policy(self, params): self._generated_password_values = [] @staticmethod - def _url_override(fields): - for parsed_field in reversed(fields or []): - if parsed_field.type != 'url': - continue - value = parsed_field.value - if value.startswith('$JSON:'): + def _record_has_primary_url(record): + if not record: + return False + if any(getattr(record, attr, None) for attr in ('link', 'login_url', 'url')): + return True + return any(getattr(field, 'type', None) == 'url' + and bool(getattr(field, 'value', None)) + for field in getattr(record, 'fields', None) or []) + + @classmethod + def _url_override(cls, fields, record=None): + url_fields = [field for field in fields or [] if field.type == 'url'] + primary_fields = [field for field in url_fields if field.section != 'c'] + selected_field = primary_fields[-1] if primary_fields else None + if selected_field is None and url_fields and not cls._record_has_primary_url(record): + selected_field = url_fields[-1] + if selected_field: + value = selected_field.value + if isinstance(value, str) and value.startswith('$JSON:'): try: value = json.loads(value[6:]) except (json.JSONDecodeError, TypeError): @@ -293,7 +306,7 @@ def _configure_password_policy(self, record, fields): password_params = getattr(self, '_password_params', None) if password_params is None: return - has_url_override, url_override = self._url_override(fields) + has_url_override, url_override = self._url_override(fields, record) kwargs = {'url_override': url_override} if has_url_override else {} self._password_policy = PasswordComplexityEnforcer.get_generator_policy( password_params, record, **kwargs) diff --git a/keepercommander/commands/recordv2.py b/keepercommander/commands/recordv2.py index 551bd6301..2af0aff14 100644 --- a/keepercommander/commands/recordv2.py +++ b/keepercommander/commands/recordv2.py @@ -33,7 +33,9 @@ add_parser.add_argument('--notes', dest='notes', action='store', help='notes') add_parser.add_argument('--custom', dest='custom', action='store', help='add custom fields. JSON or name:value pairs separated by comma. CSV Example: --custom "name1: value1, name2: value2". JSON Example: --custom \'{"name1":"value1", "name2":"value: 2,3,4"}\'') add_parser.add_argument('--folder', dest='folder', action='store', help='folder path or UID where record is to be created') -add_parser.add_argument('-f', '--force', dest='force', action='store_true', help='do not prompt for omitted fields') +add_parser.add_argument( + '-f', '--force', dest='force', action='store_true', + help='do not prompt for omitted fields and bypass password policy warnings') add_parser.add_argument('-g', '--generate', dest='generate', action='store_true', help='generate a random password') add_parser.add_argument('title', type=str, action='store', help='record title') add_parser.error = raise_parse_exception @@ -47,6 +49,9 @@ edit_parser.add_argument('--notes', dest='notes', action='store', help='set or replace the notes. Use a plus sign (+) in front appends to existing notes') edit_parser.add_argument('--custom', dest='custom', action='store', help='custom fields. JSON or name:value pairs separated by comma. CSV Example: --custom "name1: value1, name2: value2". JSON Example: --custom \'{"name1":"value1", "name2":"value: 2,3,4"}\'') edit_parser.add_argument('-g', '--generate', dest='generate', action='store_true', help='generate a random password') +edit_parser.add_argument( + '-f', '--force', dest='force', action='store_true', + help='bypass password policy warnings') edit_parser.add_argument('record', nargs='?', type=str, action='store', help='record path or UID') edit_parser.error = raise_parse_exception edit_parser.exit = suppress_exit @@ -91,8 +96,7 @@ def _enforce_password_policy(params, source, password, generated, command, force for failure in failures: logging.warning(failure) if not force: - if command == 'add': - message += ' Pass --force to bypass this policy.' + message += ' Pass --force to bypass this policy.' raise CommandError(command, message) @@ -359,10 +363,12 @@ def execute(self, params, **kwargs): if kwargs.get('password') is not None: _enforce_password_policy( - params, record, kwargs.get('password'), False, 'edit') + params, record, kwargs.get('password'), False, 'edit', + force=kwargs.get('force') is True) elif kwargs.get('generate'): _enforce_password_policy( - params, record, record.password, True, 'edit') + params, record, record.password, True, 'edit', + force=kwargs.get('force') is True) if changed: api.update_record(params, record) diff --git a/keepercommander/enforcement.py b/keepercommander/enforcement.py index 86f624842..170571eba 100644 --- a/keepercommander/enforcement.py +++ b/keepercommander/enforcement.py @@ -484,11 +484,17 @@ def _url_values(cls, source): # type: (Any) -> List[Any] value = getattr(source, attr, None) if value: urls.append(value) + if urls: + return urls if hasattr(source, 'fields') or hasattr(source, 'custom'): - for field in itertools.chain( - getattr(source, 'fields', None) or [], - getattr(source, 'custom', None) or []): + for field in getattr(source, 'fields', None) or []: + if getattr(field, 'type', None) == 'url': + value = getattr(field, 'value', None) + urls.extend(value if isinstance(value, list) else [value]) + if urls: + return urls + for field in getattr(source, 'custom', None) or []: if getattr(field, 'type', None) == 'url': value = getattr(field, 'value', None) urls.extend(value if isinstance(value, list) else [value]) @@ -504,7 +510,15 @@ def _url_values(cls, source): # type: (Any) -> List[Any] for key in ('url', 'link', 'login_url'): if data.get(key): urls.append(data[key]) - for field in itertools.chain(data.get('fields') or [], data.get('custom') or []): + if urls: + return urls + for field in data.get('fields') or []: + if isinstance(field, dict) and field.get('type') == 'url': + value = field.get('value') + urls.extend(value if isinstance(value, list) else [value]) + if urls: + return urls + for field in data.get('custom') or []: if isinstance(field, dict) and field.get('type') == 'url': value = field.get('value') urls.extend(value if isinstance(value, list) else [value]) @@ -588,15 +602,22 @@ def validate_generated_password_with_policy(cls, password, policy, algorithm): @classmethod def validate_manual_passwords(cls, params, source, passwords=None): # type: (KeeperParams, Any, Optional[List[str]]) -> List[str] - """Validate explicitly entered password values against matching domain rules only.""" + """Validate manual values against domain password rules or explicitly allowed passphrases.""" policy = cls.get_domain_specific_policy(params, source) if not policy: return [] values = passwords if passwords is not None else cls._extract_passwords(source) failures = [] for password in values: - failures.extend(cls.validate_password( - password, policy, allow_passphrase_fallback=False)) + password_failures = cls.validate_password( + password, policy, allow_passphrase_fallback=False) + if not password_failures: + continue + if policy.get('passphrase-allow') is True: + passphrase_failures = cls.validate_passphrase(password, policy) + if not passphrase_failures: + continue + failures.extend(password_failures) return failures @staticmethod diff --git a/keepercommander/generator.py b/keepercommander/generator.py index ad744b177..31bbd9467 100644 --- a/keepercommander/generator.py +++ b/keepercommander/generator.py @@ -458,11 +458,13 @@ def create_with_options(cls, policy=None, word_count=None, separator=None, cap = capitalize if cap is None: - cap = DEFAULT_PASSPHRASE_CAPITALIZE + cap = (policy.get('passphrase-capitalize', DEFAULT_PASSPHRASE_CAPITALIZE) + if policy else DEFAULT_PASSPHRASE_CAPITALIZE) num = append_number if num is None: - num = DEFAULT_PASSPHRASE_NUMBER + num = (policy.get('passphrase-number', DEFAULT_PASSPHRASE_NUMBER) + if policy else DEFAULT_PASSPHRASE_NUMBER) return cls( word_count=clamp_passphrase_word_count(wc) if isinstance(wc, int) else wc, diff --git a/unit-tests/test_nested_share_folder.py b/unit-tests/test_nested_share_folder.py index 59071e97a..8a1aacf34 100644 --- a/unit-tests/test_nested_share_folder.py +++ b/unit-tests/test_nested_share_folder.py @@ -974,6 +974,57 @@ def _side_effect(**kwargs): self.assertEqual(by_label[('checkbox', 'useSSL')]['value'], [False]) self.assertEqual(by_label[('text', 'databaseId')]['value'], ['from-sync']) + @patch('keepercommander.commands.nested_share_folder.record_commands._nsf.update_record_v3') + @patch('keepercommander.commands.nested_share_folder.helpers.check_record_edit_permission') + def test_update_retry_rechecks_manual_password_against_refreshed_url( + self, mock_perm, mock_update): + from keepercommander.commands.nested_share_folder import NestedShareRecordUpdateCommand + ruid, params = self._nsf_update_params({ + 'type': 'login', + 'title': 'Login', + 'fields': [ + {'type': 'url', 'value': ['https://other.example']}, + {'type': 'password', 'value': ['ExistingPass123']}, + ], + 'custom': [], + }) + params.enforcements = { + 'jsons': [{ + 'key': 'generated_password_complexity', + 'value': json.dumps([{ + 'domains': ['example.com'], + 'length': 12, + 'passphrase-allow': False, + }]), + }], + } + + def _side_effect(**kwargs): + if mock_update.call_count == 1: + params.record_cache[ruid]['data_unencrypted'] = json.dumps({ + 'type': 'login', + 'title': 'Login', + 'fields': [ + {'type': 'url', 'value': ['https://example.com']}, + {'type': 'password', 'value': ['ExistingPass123']}, + ], + 'custom': [], + }) + return { + 'success': False, 'status': 'RS_OUT_OF_SYNC', + 'message': 'Record changed on another client.', + } + return { + 'record_uid': ruid, 'status': 'SUCCESS', + 'message': '', 'success': True, + } + + mock_update.side_effect = _side_effect + NestedShareRecordUpdateCommand().execute( + params, record_uids=[ruid], fields=['password=weak'], force=False) + + mock_update.assert_called_once() + @patch('keepercommander.commands.nested_share_folder.record_commands._nsf.update_record_v3') @patch('keepercommander.commands.nested_share_folder.helpers.check_record_edit_permission') def test_update_sets_sync_data_after_partial_batch_failure(self, mock_perm, mock_update): diff --git a/unit-tests/test_passphrase_enforcement.py b/unit-tests/test_passphrase_enforcement.py index c23b6dc07..010741b42 100644 --- a/unit-tests/test_passphrase_enforcement.py +++ b/unit-tests/test_passphrase_enforcement.py @@ -7,6 +7,7 @@ enforce_generated_password_policy, enforce_manual_password_policy, ) +from keepercommander.commands.recordv2 import _enforce_password_policy, edit_parser from keepercommander.enforcement import PasswordComplexityEnforcer from keepercommander.error import CommandError from keepercommander import vault @@ -191,6 +192,46 @@ def test_generator_policy_merges_matching_domain_over_default(self): self.assertEqual(policy['lower-min'], 2) self.assertEqual(policy['upper-min'], 2) + def test_generated_passphrase_uses_matching_domain_generator_settings(self): + params = self._params([ + { + 'domains': ['_default_'], + 'passphrase-allow': True, + 'passphrase-length': 5, + 'passphrase-capitalize': False, + 'passphrase-number': True, + 'passphrase-separator': '-', + }, + { + 'domains': ['google.com'], + 'passphrase-allow': True, + 'passphrase-length': 7, + 'passphrase-capitalize': True, + 'passphrase-number': False, + 'passphrase-separator': '.', + }, + ]) + record = vault.TypedRecord() + record.type_name = 'login' + record.fields.extend([ + vault.TypedField.new_field('url', None, 'url'), + vault.TypedField.new_field('password', None, 'password'), + ]) + mixin = RecordEditMixin() + mixin.reset_password_policy(params) + + mixin.assign_typed_fields(record, [ + ParsedFieldValue('', 'password', '', '$GEN:passphrase'), + ParsedFieldValue('', 'url', '', 'google.com'), + ]) + + password = next(field.value[0] for field in record.fields if field.type == 'password') + words = password.split('.') + self.assertEqual(len(words), 7) + self.assertTrue(all(word[0].isupper() for word in words)) + self.assertFalse(words[0][-1].isdigit()) + self.assertEqual(mixin.warnings, []) + def test_manual_password_ignores_generator_default_policy(self): params = self._params([{ 'domains': ['_default_'], @@ -238,7 +279,7 @@ def test_manual_password_uses_matching_domain_policy(self): self.assertTrue(any('special character' in failure for failure in failures)) self.assertFalse(any('Passphrase' in failure for failure in failures)) - def test_manual_passphrase_uses_matching_domain_password_rules(self): + def test_manual_passphrase_can_satisfy_matching_domain_policy(self): params = self._params([{ 'domains': ['example.com'], 'length': 80, @@ -255,17 +296,39 @@ def test_manual_passphrase_uses_matching_domain_password_rules(self): params, self._record('https://example.com', 'alpha-bravo-charlie')) - self.assertTrue(any('at least 80 characters' in failure for failure in failures)) - self.assertTrue(any('uppercase character' in failure for failure in failures)) - self.assertFalse(any('Passphrase' in failure for failure in failures)) + self.assertEqual(failures, []) + + def test_manual_value_that_fails_both_rules_reports_password_failures(self): + params = self._params([{ + 'domains': ['example.com'], + 'length': 30, + 'upper-use': True, + 'upper-min': 1, + 'passphrase-allow': True, + 'passphrase-length': 5, + 'passphrase-capitalize': False, + 'passphrase-number': False, + 'passphrase-separator': '-', + }]) + + failures = PasswordComplexityEnforcer.validate_manual_passwords( + params, + self._record('https://example.com', 'my-secret-password')) + + self.assertTrue(any('Password must be at least 30 characters' in x for x in failures)) + self.assertTrue(any('Password must contain at least 1 uppercase' in x for x in failures)) + self.assertFalse(any('Passphrase' in x for x in failures)) - def test_passphrase_validator_accepts_unicode_words(self): + def test_manual_unicode_passphrases_use_matching_domain_passphrase_rules(self): for passphrase, capitalize in ( ('élan-café-über', False), ('Élan-Café-Über', True)): with self.subTest(passphrase=passphrase): params = self._params([{ 'domains': ['example.com'], + 'length': 80, + 'upper-use': True, + 'upper-min': 20, 'passphrase-allow': True, 'passphrase-length': 3, 'passphrase-capitalize': capitalize, @@ -273,9 +336,9 @@ def test_passphrase_validator_accepts_unicode_words(self): 'passphrase-separator': '-', }]) - policy = json.loads(params.enforcements['jsons'][0]['value'])[0] - failures = PasswordComplexityEnforcer.validate_passphrase( - passphrase, policy) + failures = PasswordComplexityEnforcer.validate_manual_passwords( + params, + self._record('https://example.com', passphrase)) self.assertEqual(failures, []) @@ -296,6 +359,86 @@ def test_manual_hyphenated_password_uses_password_rules_not_passphrase_rules(sel self.assertEqual(failures, []) + def test_manual_hyphenated_password_reports_password_failures(self): + params = self._params([{ + 'domains': ['example.com'], + 'length': 25, + 'upper-use': True, + 'upper-min': 1, + 'passphrase-allow': True, + 'passphrase-length': 5, + 'passphrase-capitalize': False, + 'passphrase-number': False, + 'passphrase-separator': '-', + }]) + + failures = PasswordComplexityEnforcer.validate_manual_passwords( + params, + self._record('https://example.com', 'my-secret-password')) + + self.assertTrue(any('at least 25 characters' in failure for failure in failures)) + self.assertTrue(any('uppercase character' in failure for failure in failures)) + self.assertFalse(any('Passphrase' in failure for failure in failures)) + + def test_primary_url_takes_precedence_over_custom_url_for_policy_selection(self): + params = self._params([ + {'domains': ['example.com'], 'length': 12}, + {'domains': ['admin.example'], 'length': 30}, + ]) + record = { + 'fields': [{'type': 'url', 'value': ['https://example.com']}], + 'custom': [{'type': 'url', 'value': ['https://admin.example']}], + } + + policy = PasswordComplexityEnforcer.get_generator_policy(params, record) + + self.assertEqual(policy['length'], 12) + + def test_custom_url_does_not_override_existing_primary_url(self): + record = vault.TypedRecord() + record.type_name = 'login' + record.fields.append(vault.TypedField.new_field( + 'url', ['https://example.com'], 'url')) + custom_url = ParsedFieldValue( + 'c', 'url', 'Admin Portal', 'https://admin.example') + + has_override, value = RecordEditMixin._url_override([custom_url], record) + + self.assertFalse(has_override) + self.assertIsNone(value) + + primary_url = ParsedFieldValue( + '', 'url', '', 'https://example.com/login') + has_override, value = RecordEditMixin._url_override( + [custom_url, primary_url], record) + + self.assertTrue(has_override) + self.assertEqual(value, 'https://example.com/login') + + empty_record = vault.TypedRecord() + empty_record.type_name = 'login' + empty_record.fields.append(vault.TypedField.new_field('url', None, 'url')) + has_override, value = RecordEditMixin._url_override([custom_url], empty_record) + + self.assertTrue(has_override) + self.assertEqual(value, 'https://admin.example') + + def test_legacy_edit_force_flag_bypasses_domain_policy(self): + args = edit_parser.parse_args(['--force', 'record-uid']) + self.assertTrue(args.force) + params = self._params([{ + 'domains': ['example.com'], + 'length': 20, + }]) + source = self._record('https://example.com') + + with self.assertRaises(CommandError) as ctx: + _enforce_password_policy( + params, source, 'weak', False, 'edit') + self.assertIn('--force', str(ctx.exception)) + + _enforce_password_policy( + params, source, 'weak', False, 'edit', force=True) def test_manual_password_with_dollar_and_digit_is_not_mistaken_for_passphrase(self): params = self._params([{ 'domains': ['google.com'], diff --git a/unit-tests/test_passphrase_generator.py b/unit-tests/test_passphrase_generator.py index 522fa5199..47a26a437 100644 --- a/unit-tests/test_passphrase_generator.py +++ b/unit-tests/test_passphrase_generator.py @@ -86,13 +86,13 @@ def test_create_with_options_overrides_policy(self): self.assertTrue(gen.capitalize) self.assertTrue(gen.append_number) - def test_commander_defaults_override_policy_capitalize_and_number(self): + def test_policy_capitalize_and_number_override_commander_defaults(self): gen = generator.KeeperPassphraseGenerator.create_with_options({ 'passphrase-capitalize': False, 'passphrase-number': False, }) - self.assertTrue(gen.capitalize) - self.assertTrue(gen.append_number) + self.assertFalse(gen.capitalize) + self.assertFalse(gen.append_number) def test_policy_separator_uses_vault_order_not_raw_first_char(self): gen = generator.KeeperPassphraseGenerator.create_with_options({ @@ -157,7 +157,7 @@ def test_record_edit_passphrase_uses_policy_when_allowed(self): with mock.patch('secrets.randbelow', return_value=4): result, error = RecordEditMixin.generate_password(['passphrase'], policy=policy) self.assertIsNone(error) - self.assertEqual(result, 'Alpha4_Bravo_Charlie_Delta_Echo') + self.assertEqual(result, 'Alpha_Bravo_Charlie_Delta_Echo') def test_record_edit_passphrase_cli_overrides_policy(self): from keepercommander.commands.record_edit import RecordEditMixin