KC-1423: fix validation for passwords and passphrase in record creation and updation, and apply domain-specific policies to manual passwords and passphrases

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,
@@ -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)
@@ -212,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)
@@ -250,7 +223,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 +252,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 +270,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)
@@ -315,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)

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,75 @@ 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 _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):
+                    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, record)
+        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 +396,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 +416,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 +689,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 +751,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 +769,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 +841,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 +1020,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 +1069,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 +1439,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 +1460,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: