KC-1423: fix validation for passwords and passphrase in record creation and updation

keepercommander/commands/nested_share_folder/record_commands.py

@@ -54,26 +54,6 @@ def _parse_field_specs(raw_fields, cmd_name):
     return record_fields, attachments
 
 
-def _apply_password_policy(editor, params, source, force):
-    # TypedRecord (typed add/update) or v3 dict (legacy add). validate_record accepts both.
-    pw_failures = PasswordComplexityEnforcer.validate_record(params, source)
-    for failure in pw_failures:
-        editor.on_warning(failure)
-    if pw_failures and not force:
-        editor.on_warning('Use --force to bypass password policy warnings.')
-
-
-def _should_stop_after_warnings(editor, force):
-    if not editor.warnings:
-        return False
-    for w in editor.warnings:
-        logging.warning(w)
-    if not force:
-        return True
-    editor.warnings.clear()
-    return False
-
-
 def _unsupported_attachment_warning(attachments, cmd_name):
     if not attachments:
         return
@@ -117,18 +97,22 @@ 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')
         folder_uid = self._resolve_folder(params, kwargs.get('folder_uid'))
 
         data = self._build_record_data(
-            params, record_type, title, notes, record_fields, kwargs.get('force'))
+            params, record_type, title, notes, record_fields)
         if self.abort_if_errors():
             return
-        if _should_stop_after_warnings(self, kwargs.get('force')):
-            return
+        self.apply_manual_password_policy_warnings(params, data)
+        if self.warnings:
+            for w in self.warnings:
+                logging.warning(w)
+            if not kwargs.get('force'):
+                return
 
         if add_attachments:
             _unsupported_attachment_warning(add_attachments, 'nsf-record-add')
@@ -156,14 +140,13 @@ def _resolve_folder(params, folder_input):
         ensure_nested_share_folder(params, uid, 'nsf-record-add', identifier=folder_input)
         return uid
 
-    def _build_record_data(self, params, record_type, title, notes, record_fields, force=False):
+    def _build_record_data(self, params, record_type, title, notes, record_fields):
         if record_type in ('legacy', 'general'):
             record = vault.PasswordRecord()
             self.assign_legacy_fields(record, record_fields)
             record.title = title
             record.notes = self.validate_notes(notes or '')
             data = self._legacy_to_data(record, title)
-            _apply_password_policy(self, params, data, force)
             return data
 
         rt_fields = self.get_record_type_fields(params, record_type)
@@ -183,7 +166,6 @@ def _build_record_data(self, params, record_type, title, notes, record_fields, f
         self.assign_typed_fields(record, record_fields)
         record.title = title
         record.notes = self.validate_notes(notes or '')
-        _apply_password_policy(self, params, record, force)
         return self._typed_to_data(record, title)
 
     @staticmethod
@@ -241,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
@@ -270,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',
@@ -286,9 +270,14 @@ def execute(self, params, **kwargs):
                     if self.abort_if_errors():
                         continue
 
-                    _apply_password_policy(self, params, record, force)
-                    if _should_stop_after_warnings(self, force):
-                        continue
+                    self.apply_manual_password_policy_warnings(params, record)
+
+                    if self.warnings:
+                        for w in self.warnings:
+                            logging.warning(w)
+                        if not force:
+                            continue
+                        self.warnings.clear()
 
                     result = self._send_typed_update(params, record_uid, record)
                     # API syncs on RS_OUT_OF_SYNC but does not retry a full data
@@ -303,9 +292,13 @@ def execute(self, params, **kwargs):
                             record_type, rt_fields, record_fields)
                         if self.abort_if_errors():
                             continue
-                        _apply_password_policy(self, params, record, force)
-                        if _should_stop_after_warnings(self, force):
-                            continue
+                        self.apply_manual_password_policy_warnings(params, record)
+                        if self.warnings:
+                            for w in self.warnings:
+                                logging.warning(w)
+                            if not force:
+                                continue
+                            self.warnings.clear()
                         result = self._send_typed_update(params, record_uid, record)
                     check_result(result, 'nsf-record-update')
                     updated += 1
@@ -350,7 +343,6 @@ def _typed_record_from_uid(self, params, record_uid):
         record = vault.TypedRecord()
         record.load_record_data(existing)
         return record
-
     @staticmethod
     def _load_record_data(params, record_uid):   # type: (Any, str) -> Optional[Dict]
         rec = params.record_cache.get(record_uid) or {}

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,6 +269,78 @@ 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."""
+        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
+        failures = PasswordComplexityEnforcer.validate_generated_password_with_policy(
+            password, password_policy, algorithm)
+        for failure in failures:
+            self.on_warning(failure)
+
     def warn_wrong_password_gen_field(self, parsed_field):
         # type: (ParsedFieldValue) -> bool
         """Warn when $GEN is used with a label like Password= instead of password=."""
@@ -321,23 +396,30 @@ 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':
                 record.login = parsed_field.value
             elif parsed_field.type == 'password':
                 action_params.clear()
                 if self.is_generate_value(parsed_field.value, action_params):
+                    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)
                     elif password is not None:
+                        self.validate_generated_password(password, algorithm)
                         record.password = password
                 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':
@@ -607,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()
@@ -666,23 +751,30 @@ 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)
                         if gen_error:
                             self.on_error(gen_error)
                             continue
+                        if value is not None:
+                            self.validate_generated_password(value, algorithm)
                     elif record_field.type in ('oneTimeCode', 'otp'):
                         value = self.generate_totp_url()
                     elif record_field.type in ('keyPair', 'privateKey'):
                         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
+                            if passphrase:
+                                self.validate_generated_password(passphrase, 'password')
                         key_type = next((x for x in action_params if x in ('rsa', 'ec', 'ed25519')), 'rsa')
                         value = self.generate_key_pair(key_type, passphrase)
                         if passphrase:
@@ -749,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)
@@ -926,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
@@ -975,18 +1069,14 @@ 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
         record.title = title
         record.notes = self.validate_notes(kwargs.get('notes') or '')
 
-        pw_failures = PasswordComplexityEnforcer.validate_record(params, record)
-        for f in pw_failures:
-            self.on_warning(f)
-        if pw_failures and not kwargs.get('force'):
-            self.on_warning('Use --force to bypass password policy warnings.')
-
         ignore_warnings = kwargs.get('force') is True
         if len(self.warnings) > 0:
             for warning in self.warnings:
@@ -1349,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.')
@@ -1370,12 +1460,7 @@ def execute(self, params, **kwargs):
         if self.abort_if_errors():
             return
 
-        if isinstance(record, vault.TypedRecord):
-            pw_failures = PasswordComplexityEnforcer.validate_record(params, record)
-            for f in pw_failures:
-                self.on_warning(f)
-            if pw_failures and not kwargs.get('force'):
-                self.on_warning('Use --force to bypass password policy warnings.')
+        self.apply_manual_password_policy_warnings(params, record)
 
         ignore_warnings = kwargs.get('force') is True
         if len(self.warnings) > 0: