diff --git a/README.md b/README.md index c9b9c49..ccbcce1 100644 --- a/README.md +++ b/README.md @@ -205,11 +205,12 @@ Current runtime ownership is intentionally narrow and explicit: tenant database. This preserves boot for tenant databases that renamed or removed the default `admin` login while still checking active default admin passwords when matching users exist. -- Startup verifies an existing administrator password before writing it. A - matching configured password is left unchanged, avoiding password-change - emails on ordinary restarts. Actual configured password changes still use - Odoo's normal write path and security notifications. Quotes and backslashes - in configured passwords are preserved when passed into the startup shell. +- Startup and post-deploy maintenance verify an existing administrator + password before writing it. A matching configured password is left + unchanged, avoiding password-change emails on ordinary restarts and deploys. + Actual configured password changes still use Odoo's normal write path and + security notifications. Quotes and backslashes in configured passwords are + preserved when passed into the Odoo shell. The same holds for a configured `ODOO_ADMIN_LOGIN` checked by the default-password policy. - A Postgres major-version bump is not a routine dependency refresh on this diff --git a/docker/scripts/run_odoo_data_workflows.py b/docker/scripts/run_odoo_data_workflows.py index f43c27a..ac4f825 100644 --- a/docker/scripts/run_odoo_data_workflows.py +++ b/docker/scripts/run_odoo_data_workflows.py @@ -1653,20 +1653,31 @@ def ensure_admin_user(self) -> None: script = textwrap.dedent(""" import json from odoo import api, SUPERUSER_ID +from odoo.exceptions import AccessDenied from odoo.modules.registry import Registry -payload = json.loads('__PAYLOAD__') +payload = json.loads(__PAYLOAD__) registry = Registry(payload['db']) with registry.cursor() as cr: env = api.Environment(cr, SUPERUSER_ID, {}) admin = env['res.users'].sudo().search([('login', '=', payload['login'])], limit=1) if admin: if payload['set_password']: - admin.with_context(no_reset_password=True).sudo().write({'password': payload['password']}) + # Odoo emails a security notice on every password write, so skip unchanged passwords. + try: + admin.with_user(admin)._check_credentials( + {'type': 'password', 'password': payload['password']}, + {'interactive': True}, + ) + except AccessDenied: + admin.with_context(no_reset_password=True).sudo().write({'password': payload['password']}) + print('admin_password_updated=true') + else: + print('admin_password_updated=false') if payload['set_email'] and admin.partner_id: admin.partner_id.sudo().write({'email': payload['email']}) cr.commit() -""").replace("__PAYLOAD__", json.dumps(payload)) +""").replace("__PAYLOAD__", repr(json.dumps(payload))) _logger.info("Applying admin hardening updates.") self._run_odoo_shell(script, "admin hardening") diff --git a/tests/test_odoo_data_workflows.py b/tests/test_odoo_data_workflows.py index da23603..dd0c48b 100644 --- a/tests/test_odoo_data_workflows.py +++ b/tests/test_odoo_data_workflows.py @@ -1350,5 +1350,88 @@ def test_explicit_modules_override_configured_modules(self) -> None: self.assertEqual(list(apply_module_updates.call_args.args[0]), ["website"]) +class EnsureAdminUserTests(unittest.TestCase): + def _runner(self, admin_password: str) -> Any: + environment: dict[str, object] = { + "ODOO_DB_HOST": "database", + "ODOO_DB_USER": "odoo", + "ODOO_DB_PASSWORD": "database-password", + "ODOO_DB_NAME": "cm_website", + "ODOO_FILESTORE_PATH": "/volumes/data/filestore/cm_website", + "ODOO_ADMIN_PASSWORD": admin_password, + "PLATFORM_INSTANCE": "testing", + } + with patch.dict(os.environ, {}, clear=True): + settings = odoo_data_workflows.LocalServerSettings(**environment) + runner = odoo_data_workflows.OdooDataWorkflowRunner(settings, upstream=None, env_file=None) + cursor = MagicMock() + cursor.fetchone.side_effect = lambda: (2, 3) if "res_users" in cursor.execute.call_args.args[0] else ("admin@localhost",) + runner.local.db_conn = MagicMock() + runner.local.db_conn.cursor.return_value.__enter__.return_value = cursor + for name in ("connect_to_db", "_reset_db_connection"): + patcher = patch.object(runner, name) + patcher.start() + self.addCleanup(patcher.stop) + return runner + + @staticmethod + def _run_admin_hardening(runner: Any, environment: MagicMock) -> None: + odoo_module = types.ModuleType("odoo") + odoo_module.__dict__.update(api=MagicMock(Environment=MagicMock(return_value=environment)), SUPERUSER_ID=1) + exceptions = types.ModuleType("odoo.exceptions") + exceptions.__dict__["AccessDenied"] = PermissionError + registry_module = types.ModuleType("odoo.modules.registry") + registry_module.__dict__["Registry"] = MagicMock() + + def run_shell(script: str, label: str) -> None: + if label == "admin hardening": + exec(script, {}) + + with ( + patch.dict( + sys.modules, + {"odoo": odoo_module, "odoo.exceptions": exceptions, "odoo.modules.registry": registry_module}, + ), + patch.object(runner, "_run_odoo_shell", side_effect=run_shell), + ): + runner.ensure_admin_user() + + def test_post_deploy_admin_hardening_only_writes_when_configured_password_changes(self) -> None: + environment = MagicMock() + admin = environment["res.users"].sudo().search() + admin.with_user.return_value = admin + admin.with_context.return_value = admin + admin.sudo.return_value = admin + stored = {"password": "initial-password"} + + def check_credentials(credential: dict[str, str], _request_environment: dict[str, bool]) -> None: + if credential["password"] != stored["password"]: + raise PermissionError + + admin._check_credentials.side_effect = check_credentials + admin.write.side_effect = stored.update + configured_password = "configured-'\"\\-password" + + self._run_admin_hardening(self._runner(configured_password), environment) + self._run_admin_hardening(self._runner(configured_password), environment) + admin.write.assert_called_once_with({"password": configured_password}) + + self._run_admin_hardening(self._runner("rotated-password"), environment) + self._run_admin_hardening(self._runner("rotated-password"), environment) + self.assertEqual(admin.write.call_count, 2) + self.assertEqual(stored["password"], "rotated-password") + + def test_post_deploy_admin_hardening_does_not_write_after_unexpected_credential_check_failure(self) -> None: + environment = MagicMock() + admin = environment["res.users"].sudo().search() + admin.with_user.return_value = admin + admin._check_credentials.side_effect = RuntimeError("credential backend unavailable") + + with self.assertRaisesRegex(RuntimeError, "credential backend unavailable"): + self._run_admin_hardening(self._runner("configured-password"), environment) + + admin.with_context.assert_not_called() + + if __name__ == "__main__": unittest.main()