From 2870b7ea16dc9bd473c8a01f77b1304ab577cf5a Mon Sep 17 00:00:00 2001
From: sshrushanth-ks <sshrushanth@keepersecurity.com>
Date: Fri, 28 Aug 2026 15:32:19 +0530
Subject: [PATCH 1/5] KC-1423: fix validation for passwords and passphrase in
 record creation and updation (#2314)

* fix validation for passwords and passphrase in record creation and updation

* Fix generated password policy validation after review

- Track generated key-pair passphrases
- Centralize v3 add/edit policy validation
- Preserve random-password diagnostics
- Keep manual passwords exempt from generated-password policy
- Add regression coverage across record and NSF commands

* Validate generated passwords at generation site instead of post-hoc via
instance flags. This fixes two issues:

1. v3 add/edit --generate now validates (was silently bypassed)
2. No cross-field contamination in multi-field records

Manual passwords remain unvalidated (handled by BreachWatch).

* updated failing test
---
 .../nested_share_folder/record_commands.py    | 31 +++++++++-
 keepercommander/commands/record_edit.py       | 30 ++++++----
 keepercommander/commands/recordv3.py          | 44 +++++++-------
 keepercommander/enforcement.py                |  8 +--
 unit-tests/test_nested_share_folder.py        | 16 +++--
 unit-tests/test_passphrase_enforcement.py     | 60 +++++++++++++++++++
 unit-tests/test_passphrase_generator.py       | 16 +++++
 7 files changed, 162 insertions(+), 43 deletions(-)

diff --git a/keepercommander/commands/nested_share_folder/record_commands.py b/keepercommander/commands/nested_share_folder/record_commands.py
index faf277ce1..1671bfb39 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 nested_share_folder as _nsf, vault, vault_extensions
+from ... import generator, 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,
@@ -231,6 +231,34 @@ 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)
@@ -350,7 +378,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 {}
diff --git a/keepercommander/commands/record_edit.py b/keepercommander/commands/record_edit.py
index 1560a6e46..826e58947 100644
--- a/keepercommander/commands/record_edit.py
+++ b/keepercommander/commands/record_edit.py
@@ -266,6 +266,16 @@ def abort_if_errors(self):
     def on_info(self, message):
         logging.info(message)
 
+    def validate_generated_password(self, password, algorithm):
+        """Validate a generated password with the correct algorithm flag."""
+        if not password or not self._password_policy:
+            return
+        allow_passphrase = algorithm == 'passphrase'
+        failures = PasswordComplexityEnforcer.validate_password(
+            password, self._password_policy, allow_passphrase_fallback=allow_passphrase)
+        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=."""
@@ -328,10 +338,12 @@ def assign_legacy_fields(self, record, fields):
             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:
@@ -669,10 +681,13 @@ def assign_typed_fields(self, record, fields):
                 action_params = []
                 if self.is_generate_value(parsed_field.value, action_params):
                     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'):
@@ -683,6 +698,8 @@ def assign_typed_fields(self, record, fields):
                             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:
@@ -981,12 +998,6 @@ def execute(self, params, **kwargs):
         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:
@@ -1370,13 +1381,6 @@ 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.')
-
         ignore_warnings = kwargs.get('force') is True
         if len(self.warnings) > 0:
             for warning in self.warnings:
diff --git a/keepercommander/commands/recordv3.py b/keepercommander/commands/recordv3.py
index f286800b6..b55d55a70 100644
--- a/keepercommander/commands/recordv3.py
+++ b/keepercommander/commands/recordv3.py
@@ -142,6 +142,22 @@ def get_password_from_rules(generate_rules, generate_length):
     return kpg.generate()
 
 
+def enforce_generated_password_policy(params, data, command, generated, manual_password=None, force=False):
+    # 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)
+    if pw_failures:
+        for failure in pw_failures:
+            logging.warning(bcolors.WARNING + failure + bcolors.ENDC)
+        if not force:
+            raise CommandError(
+                command,
+                'Password does not meet enterprise complexity policy. '
+                'Pass --force to bypass these warnings.')
+
+
 class RecordAddCommand(Command, recordv2.RecordUtils):
     def get_parser(self):
         return add_parser
@@ -404,21 +420,15 @@ def GCM_TAG_LEN(): return 16
 
         # For compatibility w/ legacy: --password overrides --generate AND --generate overrides dataJSON/option
         # dataJSON/option < kwargs: --generate < kwargs: --password
-        password = kwargs.get('password')
+        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'))
         if password:
             data = recordv3.RecordV3.update_password(password, data, recordv3.RecordV3.get_record_type_definition(params, data))
 
-        pw_failures = PasswordComplexityEnforcer.validate_record(params, data)
-        if pw_failures:
-            for f in pw_failures:
-                logging.warning(bcolors.WARNING + f + bcolors.ENDC)
-            if not kwargs.get('force'):
-                raise CommandError(
-                    'add',
-                    'Password does not meet enterprise complexity policy. '
-                    'Pass --force to bypass these warnings.')
+        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'))
 
         record_uid = api.generate_record_uid()
         logging.debug('Generated Record UID: %s', record_uid)
@@ -653,22 +663,16 @@ def execute(self, params, **kwargs):
 
         # For compatibility w/ legacy: --password overides --generate AND --generate overrides dataJSON/option
         # dataJSON/option < kwargs: --generate < kwargs: --password
-        password = kwargs.get('password')
+        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'))
         if password:
             record.password = password
             data = recordv3.RecordV3.update_password(password, data, recordv3.RecordV3.get_record_type_definition(params, data))
 
-        pw_failures = PasswordComplexityEnforcer.validate_record(params, data)
-        if pw_failures:
-            for f in pw_failures:
-                logging.warning(bcolors.WARNING + f + bcolors.ENDC)
-            if not kwargs.get('force'):
-                raise CommandError(
-                    'edit',
-                    'Password does not meet enterprise complexity policy. '
-                    'Pass --force to bypass these warnings.')
+        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'))
 
         data_dict = json.loads(data)
         changed = rdata_dict != data_dict
diff --git a/keepercommander/enforcement.py b/keepercommander/enforcement.py
index 2e3cd0a43..a53e824eb 100644
--- a/keepercommander/enforcement.py
+++ b/keepercommander/enforcement.py
@@ -505,7 +505,7 @@ def validate_passphrase(cls, password, policy):   # type: (str, Dict[str, Any])
         return failures
 
     @classmethod
-    def validate_password(cls, password, policy):   # type: (str, Dict[str, Any]) -> List[str]
+    def validate_password(cls, password, policy, allow_passphrase_fallback=True):   # type: (str, Dict[str, Any], bool) -> List[str]
         failures = []   # type: List[str]
         if not policy or not isinstance(password, str) or not password:
             return failures
@@ -540,7 +540,7 @@ def validate_password(cls, password, policy):   # type: (str, Dict[str, Any]) ->
             return []
 
         # Vault re-validates as a passphrase when random password rules fail.
-        if policy.get('passphrase-allow') is False:
+        if not allow_passphrase_fallback or policy.get('passphrase-allow') is False:
             return failures
 
         passphrase_failures = cls.validate_passphrase(password, policy)
@@ -550,7 +550,7 @@ def validate_password(cls, password, policy):   # type: (str, Dict[str, Any]) ->
         return passphrase_failures
 
     @classmethod
-    def validate_record(cls, params, source):   # type: (KeeperParams, Any) -> List[str]
+    def validate_record(cls, params, source, allow_passphrase_fallback=True):   # type: (KeeperParams, Any, bool) -> List[str]
         """Return policy violations across all password fields in `source`.
 
         `source` may be a vault.TypedRecord, a v3 record-data dict, or a JSON
@@ -562,7 +562,7 @@ def validate_record(cls, params, source):   # type: (KeeperParams, Any) -> List[
             return []
         failures = []   # type: List[str]
         for pw in cls._extract_passwords(source):
-            failures.extend(cls.validate_password(pw, policy))
+            failures.extend(cls.validate_password(pw, policy, allow_passphrase_fallback))
         return failures
 
     @staticmethod
diff --git a/unit-tests/test_nested_share_folder.py b/unit-tests/test_nested_share_folder.py
index 0c3fac318..82621369c 100644
--- a/unit-tests/test_nested_share_folder.py
+++ b/unit-tests/test_nested_share_folder.py
@@ -492,8 +492,12 @@ def test_add_record_rejects_restricted_record_type(self, mock_create):
         mock_create.assert_not_called()
 
     @patch('keepercommander.commands.nested_share_folder.record_commands._nsf.create_record_v3')
-    def test_add_record_rejects_weak_password_without_force(self, mock_create):
+    def test_add_record_allows_manual_weak_password(self, mock_create):
         from keepercommander.commands.nested_share_folder import NestedShareRecordAddCommand
+        mock_create.return_value = {
+            'record_uid': utils.generate_uid(), 'status': 'SUCCESS',
+            'message': '', 'success': True, 'revision': 1,
+        }
         fuid, fobj = _make_folder()
         params = _make_params(
             nested_share_folders={fuid: fobj},
@@ -512,7 +516,7 @@ def test_add_record_rejects_weak_password_without_force(self, mock_create):
         cmd = NestedShareRecordAddCommand()
         cmd.execute(params, title='Weak', record_type='general',
                     fields=['password=abc'], force=False)
-        mock_create.assert_not_called()
+        mock_create.assert_called_once()
 
     @patch('keepercommander.commands.nested_share_folder.record_commands._nsf.create_record_v3')
     def test_add_record_allows_weak_password_with_force(self, mock_create):
@@ -603,8 +607,12 @@ def test_update_record_rejects_restricted_record_type(self, mock_perm, mock_upda
 
     @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_rejects_weak_password_without_force(self, mock_perm, mock_update):
+    def test_update_record_allows_manual_weak_password(self, mock_perm, mock_update):
         from keepercommander.commands.nested_share_folder import NestedShareRecordUpdateCommand
+        mock_update.return_value = {
+            'record_uid': 'x', 'status': 'SUCCESS',
+            'message': '', 'success': True, 'revision': 2,
+        }
         ruid, robj = _make_record()
         params = _make_params(
             nested_share_records={ruid: robj},
@@ -626,7 +634,7 @@ def test_update_record_rejects_weak_password_without_force(self, mock_perm, mock
         )
         cmd = NestedShareRecordUpdateCommand()
         cmd.execute(params, record_uids=[ruid], fields=['password=abc'], force=False)
-        mock_update.assert_not_called()
+        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')
diff --git a/unit-tests/test_passphrase_enforcement.py b/unit-tests/test_passphrase_enforcement.py
index 4093ba7de..9bc0f6471 100644
--- a/unit-tests/test_passphrase_enforcement.py
+++ b/unit-tests/test_passphrase_enforcement.py
@@ -1,6 +1,10 @@
+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.enforcement import PasswordComplexityEnforcer
+from keepercommander.error import CommandError
 
 
 STRICT_RANDOM_POLICY = {
@@ -82,3 +86,59 @@ def test_random_password_still_validated(self):
         password = 'ABCDE123!!!'
         failures = PasswordComplexityEnforcer.validate_password(password, STRICT_RANDOM_POLICY)
         self.assertEqual(failures, [])
+
+    def test_invalid_random_password_keeps_random_policy_errors(self):
+        policy = dict(STRICT_RANDOM_POLICY)
+        policy.update({'length': 20, 'passphrase-length': 5, 'passphrase-separator': '!'})
+        failures = PasswordComplexityEnforcer.validate_password(
+            'ABC123!!', policy, allow_passphrase_fallback=False)
+        self.assertTrue(any('Password must be at least 20 characters' in f for f in failures))
+        self.assertFalse(any('Passphrase contains' in f for f in failures))
+
+
+class TestGeneratedPasswordPolicyWarnings(TestCase):
+
+    def _params(self):
+        return SimpleNamespace(enforcements={
+            'jsons': [{
+                'key': 'generated_password_complexity',
+                'value': '{"length": 20, "passphrase-allow": true, "passphrase-length": 5}',
+            }],
+        })
+
+    def test_generated_random_password_fails_when_too_short(self):
+        mixin = RecordEditMixin()
+        mixin._password_policy = PasswordComplexityEnforcer.get_policy(self._params())
+        mixin.validate_generated_password('pass', 'password')
+        self.assertTrue(any('Password must be at least' in w for w in mixin.warnings))
+
+    def test_generated_passphrase_validates_with_passphrase_fallback(self):
+        mixin = RecordEditMixin()
+        mixin._password_policy = PasswordComplexityEnforcer.get_policy(self._params())
+        mixin.validate_generated_password('pass', 'passphrase')
+        # Passphrase validation is looser, may not fail on short input
+        self.assertTrue(isinstance(mixin.warnings, list))
+
+
+class TestV3GeneratedPasswordPolicy(TestCase):
+
+    def _params(self):
+        return SimpleNamespace(enforcements={
+            'jsons': [{
+                'key': 'generated_password_complexity',
+                'value': '{"length": 20}',
+            }],
+        })
+
+    def _record(self):
+        return {'fields': [{'type': 'password', 'value': ['pass']}]}
+
+    def test_generated_password_is_rejected_when_policy_fails(self):
+        with self.assertRaises(CommandError):
+            enforce_generated_password_policy(
+                self._params(), self._record(), 'add', generated=True, manual_password=None)
+
+    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)
diff --git a/unit-tests/test_passphrase_generator.py b/unit-tests/test_passphrase_generator.py
index 8cf9b40e2..522fa5199 100644
--- a/unit-tests/test_passphrase_generator.py
+++ b/unit-tests/test_passphrase_generator.py
@@ -205,6 +205,22 @@ def test_invalid_gen_algorithm_aborts_record_add(self):
     self.assertEqual(len(cmd.errors), 1)
     self.assertIn('passphrase', cmd.errors[0])
 
+  def test_encrypted_key_pair_validates_passphrase(self):
+    from keepercommander.commands.record_edit import RecordAddCommand, ParsedFieldValue
+    from keepercommander import vault
+    cmd = RecordAddCommand()
+    record = vault.TypedRecord()
+    record.type_name = 'sshKeys'
+    record.fields.append(vault.TypedField.new_field('keyPair', '', ''))
+    with mock.patch.object(cmd, 'generate_password', return_value=('weak', None)), \
+            mock.patch.object(cmd, 'generate_key_pair', return_value={'privateKey': 'key'}), \
+            mock.patch.object(cmd, 'validate_generated_password') as mock_validate:
+      cmd.assign_typed_fields(record, [
+        ParsedFieldValue('', 'keyPair', '', '$GEN:enc'),
+      ])
+      # Verify validate_generated_password was called for the passphrase
+      mock_validate.assert_called_once_with('weak', 'password')
+
   def test_invalid_passphrase_separator_aborts_generation(self):
     from keepercommander.commands.record_edit import RecordEditMixin
     result, error = RecordEditMixin.generate_password(

From e2dd1f906ec9169771021aa38a7da71ea97dabe3 Mon Sep 17 00:00:00 2001
From: sshrushanth-ks <sshrushanth@keepersecurity.com>
Date: Fri, 28 Aug 2026 16:17:59 +0530
Subject: [PATCH 2/5] updated record_commands

---
 .../nested_share_folder/record_commands.py    | 51 +++++++------------
 1 file changed, 19 insertions(+), 32 deletions(-)

diff --git a/keepercommander/commands/nested_share_folder/record_commands.py b/keepercommander/commands/nested_share_folder/record_commands.py
index 1671bfb39..4748ac7f7 100644
--- a/keepercommander/commands/nested_share_folder/record_commands.py
+++ b/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
@@ -124,11 +104,14 @@ def execute(self, params, **kwargs):
         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
+        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 +139,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 +165,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
@@ -314,9 +295,12 @@ 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
+                    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
@@ -331,9 +315,12 @@ 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
+                        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

From 8abbb9074f90c426725ca93c3d595cf4d15e5c19 Mon Sep 17 00:00:00 2001
From: sshrushanth-ks <sshrushanth@keepersecurity.com>
Date: Fri, 28 Aug 2026 16:33:21 +0530
Subject: [PATCH 3/5] updated test file

---
 unit-tests/test_nested_share_folder.py | 14 +++++++++-----
 1 file changed, 9 insertions(+), 5 deletions(-)

diff --git a/unit-tests/test_nested_share_folder.py b/unit-tests/test_nested_share_folder.py
index 82621369c..da233821a 100644
--- a/unit-tests/test_nested_share_folder.py
+++ b/unit-tests/test_nested_share_folder.py
@@ -973,7 +973,7 @@ def test_update_sets_sync_data_after_partial_batch_failure(self, mock_perm, mock
 
     @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_skips_warned_record_and_continues_batch(self, mock_perm, mock_update):
+    def test_update_allows_weak_manual_passwords_in_batch(self, mock_perm, mock_update):
         from keepercommander.commands.nested_share_folder import NestedShareRecordUpdateCommand
         policy = json.dumps([{
             'length': 12,
@@ -996,11 +996,11 @@ def test_update_skips_warned_record_and_continues_batch(self, mock_perm, mock_up
             }),
         }
         params.enforcements = {'jsons': [{'key': 'generated_password_complexity', 'value': policy}]}
+        self._stub_update(mock_update, ruid1)
         self._stub_update(mock_update, ruid2)
         NestedShareRecordUpdateCommand().execute(
             params, record_uids=[ruid1, ruid2], title='Updated', fields=[], force=False)
-        self.assertEqual(mock_update.call_count, 1)
-        self.assertEqual(mock_update.call_args.kwargs['record_uid'], ruid2)
+        self.assertEqual(mock_update.call_count, 2)
         self.assertTrue(params.sync_data)
 
     @patch('keepercommander.commands.nested_share_folder.record_commands._nsf.update_record_v3')
@@ -1033,8 +1033,12 @@ def test_update_empty_file_spec_warns_as_removal(self, mock_perm, mock_update):
         mock_update.assert_not_called()
 
     @patch('keepercommander.commands.nested_share_folder.record_commands._nsf.create_record_v3')
-    def test_add_rejects_weak_custom_password_without_force(self, mock_create):
+    def test_add_allows_weak_custom_password_without_force(self, mock_create):
         from keepercommander.commands.nested_share_folder import NestedShareRecordAddCommand
+        mock_create.return_value = {
+            'record_uid': utils.generate_uid(), 'status': 'SUCCESS',
+            'message': '', 'success': True, 'revision': 1,
+        }
         fuid, fobj = _make_folder()
         params = _make_params(
             nested_share_folders={fuid: fobj},
@@ -1055,7 +1059,7 @@ def test_add_rejects_weak_custom_password_without_force(self, mock_create):
                                return_value=[{'$ref': 'login'}, {'$ref': 'password'}]):
             cmd.execute(params, title='Weak custom', record_type='login',
                         fields=['c.password.AppSecret=abc'], force=False)
-        mock_create.assert_not_called()
+        mock_create.assert_called_once()
 
     @patch('keepercommander.commands.nested_share_folder.record_commands._nsf.create_record_v3')
     def test_add_unescapes_notes(self, mock_create):

From b2109a6f7cf9ac4b712fba9781d70aaa3ef88983 Mon Sep 17 00:00:00 2001
From: sshrushanth-ks <sshrushanth@keepersecurity.com>
Date: Fri, 28 Aug 2026 16:41:06 +0530
Subject: [PATCH 4/5] updated nsf test file

---
 unit-tests/test_nested_share_folder.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/unit-tests/test_nested_share_folder.py b/unit-tests/test_nested_share_folder.py
index da233821a..5e5116a40 100644
--- a/unit-tests/test_nested_share_folder.py
+++ b/unit-tests/test_nested_share_folder.py
@@ -460,7 +460,7 @@ def setUp(self):
     def tearDown(self):
         mock.patch.stopall()
 
-    @patch('keepercommander.nested_share_folder.record_api.create_record_v3')
+    @patch('keepercommander.commands.nested_share_folder.record_commands._nsf.create_record_v3')
     def test_add_record(self, mock_create):
         from keepercommander.commands.nested_share_folder import NestedShareRecordAddCommand
         mock_create.return_value = {

From e0fd975b735308ef40b75085e0f5a3ce9c4b1a6d Mon Sep 17 00:00:00 2001
From: sshrushanth-ks <sshrushanth@keepersecurity.com>
Date: Fri, 30 Oct 2026 18:59:10 +0530
Subject: [PATCH 5/5] KC-1423: fix validation for passwords and passphrase in
 record creation and updation, and apply domain-specific policies to manual
 passwords and passphrases (#2453)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

* fix validation for passwords and passphrase in record creation and updation

* 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.

* 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.

* 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.

* Addressed review comments:

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    |  40 +-
 keepercommander/commands/record_edit.py       |  95 +++-
 keepercommander/commands/recordv2.py          |  57 ++-
 keepercommander/commands/recordv3.py          |  51 ++-
 keepercommander/enforcement.py                | 244 +++++++++-
 keepercommander/generator.py                  |   6 +-
 unit-tests/test_nested_share_folder.py        |  83 +++-
 unit-tests/test_passphrase_enforcement.py     | 418 +++++++++++++++++-
 unit-tests/test_passphrase_generator.py       |   8 +-
 9 files changed, 920 insertions(+), 82 deletions(-)

diff --git a/keepercommander/commands/nested_share_folder/record_commands.py b/keepercommander/commands/nested_share_folder/record_commands.py
index 4748ac7f7..1fc8c64c6 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,
@@ -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)
diff --git a/keepercommander/commands/record_edit.py b/keepercommander/commands/record_edit.py
index 826e58947..0b6c523f6 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,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:
diff --git a/keepercommander/commands/recordv2.py b/keepercommander/commands/recordv2.py
index e382b774e..2af0aff14 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
@@ -32,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
@@ -46,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
@@ -76,6 +82,24 @@ 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:
+            message += ' Pass --force to bypass this policy.'
+            raise CommandError(command, message)
+
+
 class RecordAddCommand(Command, RecordUtils):
     def get_parser(self):
         return add_parser
@@ -90,7 +114,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 +122,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 +299,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 +361,15 @@ 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',
+                force=kwargs.get('force') is True)
+        elif kwargs.get('generate'):
+            _enforce_password_policy(
+                params, record, record.password, True, 'edit',
+                force=kwargs.get('force') is True)
+
         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..170571eba 100644
--- a/keepercommander/enforcement.py
+++ b/keepercommander/enforcement.py
@@ -15,8 +15,10 @@
 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
 
 from . import api, utils, crypto
 from .proto import APIRequest_pb2, record_pb2
@@ -67,6 +69,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 +417,215 @@ 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 urls:
+            return urls
+
+        if hasattr(source, 'fields') or hasattr(source, 'custom'):
+            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])
+            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])
+            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])
+        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 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:
+            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
+    def _passphrase_word_letters(word):
+        # type: (str) -> Optional[str]
+        word = unicodedata.normalize('NFC', word)
+        return word if len(word) >= 3 and word.isalpha() else None
+
     @classmethod
     def _normalize_passphrase_separators(cls, policy):   # type: (Dict[str, Any]) -> Optional[str]
         raw = policy.get('passphrase-separator')
@@ -477,22 +674,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/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 5e5116a40..8a1aacf34 100644
--- a/unit-tests/test_nested_share_folder.py
+++ b/unit-tests/test_nested_share_folder.py
@@ -636,6 +636,37 @@ def test_update_record_allows_manual_weak_password(self, mock_perm, mock_update)
         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):
@@ -837,7 +868,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()
@@ -944,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 9bc0f6471..010741b42 100644
--- a/unit-tests/test_passphrase_enforcement.py
+++ b/unit-tests/test_passphrase_enforcement.py
@@ -1,10 +1,16 @@
+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.commands.recordv2 import _enforce_password_policy, edit_parser
 from keepercommander.enforcement import PasswordComplexityEnforcer
 from keepercommander.error import CommandError
+from keepercommander import vault
 
 
 STRICT_RANDOM_POLICY = {
@@ -142,3 +148,411 @@ 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_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_'],
+            '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_can_satisfy_matching_domain_policy(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_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_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,
+                    'passphrase-number': False,
+                    'passphrase-separator': '-',
+                }])
+
+                failures = PasswordComplexityEnforcer.validate_manual_passwords(
+                    params,
+                    self._record('https://example.com', passphrase))
+
+                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_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'],
+            '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'])
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