diff --git a/vault/models/__init__.py b/vault/models/__init__.py index d9d3e40ebe..8879dfbdb7 100644 --- a/vault/models/__init__.py +++ b/vault/models/__init__.py @@ -4,6 +4,7 @@ from . import ( abstract_vault, abstract_vault_field, + res_company, res_config_settings, res_users, res_users_key, diff --git a/vault/models/res_company.py b/vault/models/res_company.py new file mode 100644 index 0000000000..31e1c43b16 --- /dev/null +++ b/vault/models/res_company.py @@ -0,0 +1,37 @@ +# © 2026 Nitrokey GmbH +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl). + +from odoo import _, api, fields, models +from odoo.exceptions import ValidationError + + +class ResCompany(models.Model): + _inherit = "res.company" + + vault_custodian_ids = fields.Many2many( + "res.users", + "vault_company_custodian_rel", + "company_id", + "user_id", + string="Vault Custodians", + domain=[("has_vault_key", "=", True)], + help="Users configured here are automatically added to every newly " + "created vault and can not be removed from it. They keep access to the " + "end-to-end encrypted vaults, for example to recover the data in case " + "an employee leaves the company.", + ) + + @api.constrains("vault_custodian_ids") + def _check_vault_custodian_keys(self): + for company in self: + keyless = company.vault_custodian_ids.filtered( + lambda u: not u.has_vault_key + ) + if keyless: + raise ValidationError( + _( + "The following users can not be vault custodians because " + "they have no vault keys: %s" + ) + % ", ".join(keyless.mapped("display_name")) + ) diff --git a/vault/models/res_config_settings.py b/vault/models/res_config_settings.py index 105571263f..5955601590 100644 --- a/vault/models/res_config_settings.py +++ b/vault/models/res_config_settings.py @@ -14,3 +14,8 @@ class ResConfigSettings(models.TransientModel): group_vault_import = fields.Boolean( "Import Vaults", implied_group="vault.group_vault_import" ) + vault_custodian_ids = fields.Many2many( + related="company_id.vault_custodian_ids", + string="Mandatory Custodians", + readonly=False, + ) diff --git a/vault/models/res_users.py b/vault/models/res_users.py index 8c6c75e231..c3c19db92e 100644 --- a/vault/models/res_users.py +++ b/vault/models/res_users.py @@ -18,12 +18,31 @@ class ResUsers(models.Model): store=False, ) keys = fields.One2many("res.users.key", "user_id", readonly=True) + has_vault_key = fields.Boolean( + compute="_compute_has_vault_key", + search="_search_has_vault_key", + help="Whether the user has vault keys configured", + ) vault_right_ids = fields.One2many("vault.right", "user_id", readonly=True) inbox_ids = fields.One2many("vault.inbox", "user_id") inbox_enabled = fields.Boolean(default=True) inbox_link = fields.Char(compute="_compute_inbox_link", readonly=True, store=False) inbox_token = fields.Char(default=lambda self: uuid4(), readonly=True) + @api.depends("keys") + def _compute_has_vault_key(self): + for rec in self: + rec.has_vault_key = bool(rec.sudo().keys) + + @api.model + def _search_has_vault_key(self, operator, value): + if operator not in ("=", "!=") or not isinstance(value, bool): + raise ValueError(self.env._("Unsupported search operator")) + + users_with_keys = self.env["res.users.key"].sudo().search([]).mapped("user_id") + has_key = (operator == "=") == value + return [("id", "in" if has_key else "not in", users_with_keys.ids)] + @api.depends("keys", "keys.current") def _compute_active_key(self): for rec in self: diff --git a/vault/models/vault.py b/vault/models/vault.py index 59eb5c1698..0dc28680f7 100644 --- a/vault/models/vault.py +++ b/vault/models/vault.py @@ -5,6 +5,7 @@ from uuid import uuid4 from odoo import _, api, fields, models +from odoo.exceptions import UserError _logger = logging.getLogger(__name__) @@ -100,7 +101,7 @@ def _inverse_master_key(self): rights.key = rec.master_key def _get_default_rights(self): - return [ + rights = [ ( 0, 0, @@ -114,6 +115,63 @@ def _get_default_rights(self): ) ] + custodians = self.env.company.sudo().vault_custodian_ids + for custodian in custodians: + if custodian.id == self.env.uid: + continue + + rights.append( + ( + 0, + 0, + { + "user_id": custodian.id, + "perm_create": False, + "perm_write": False, + "perm_delete": False, + "perm_share": True, + }, + ) + ) + + return rights + + @api.model_create_multi + def create(self, vals_list): + records = super().create(vals_list) + records._enforce_custodians() + return records + + def _enforce_custodians(self): + custodians = self.env.company.sudo().vault_custodian_ids + if not custodians: + return + + required = custodians.filtered(lambda u: u.id != self.env.uid) + for rec in self: + custodian_rights = rec.right_ids.filtered( + lambda r, required=required: r.user_id in required + ) + missing = required - custodian_rights.user_id + if missing: + raise UserError( + _( + "The following mandatory vault custodians must be shared " + "with the vault and can not be removed: %s" + ) + % ", ".join(missing.mapped("display_name")) + ) + + without_share = custodian_rights.filtered(lambda r: not r.perm_share) + if without_share: + raise UserError( + _( + "The following mandatory vault custodians must have the " + "share permission: %s" + ) + % ", ".join(without_share.user_id.mapped("display_name")) + ) + def _log_entry(self, msg, state): self.ensure_one() return ( diff --git a/vault/models/vault_right.py b/vault/models/vault_right.py index 6e90338674..0fec5c69d1 100644 --- a/vault/models/vault_right.py +++ b/vault/models/vault_right.py @@ -1,7 +1,8 @@ # © 2021 Florian Kantelberg - initOS GmbH # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl). -from odoo import api, fields, models +from odoo import _, api, fields, models +from odoo.exceptions import UserError class VaultRight(models.Model): @@ -63,6 +64,10 @@ class VaultRight(models.Model): def _get_is_owner(self): return self.env.user == self.vault_id.user_id + def _filtered_custodians(self): + custodians = self.env.company.sudo().vault_custodian_ids + return self.filtered(lambda r: r.user_id in custodians) + @api.depends("user_id") def _compute_public_key(self): for rec in self: @@ -95,6 +100,15 @@ def create(self, vals_list): return res def write(self, values): + # Prevent revoking the share with a mandatory custodian + if not self.env.su and self._filtered_custodians(): + if values.get("perm_share") is False: + raise UserError( + _("The share permission of a custodian can not be removed.") + ) + if "user_id" in values: + raise UserError(_("The user of a custodian can not be changed.")) + res = super().write(values) perms = ["perm_write", "perm_delete", "perm_share", "perm_create"] if any(x in values for x in perms): @@ -103,6 +117,11 @@ def write(self, values): return res def unlink(self): + if not self.env.su and self._filtered_custodians(): + raise UserError( + _("A mandatory custodian can not be removed from the vault.") + ) + for rec in self: rec.vault_id.log_info(f"Removed user {self.user_id.display_name}") rec.vault_id.reencrypt_required = True diff --git a/vault/readme/DESCRIPTION.md b/vault/readme/DESCRIPTION.md index 32eb5769c4..47509b362c 100644 --- a/vault/readme/DESCRIPTION.md +++ b/vault/readme/DESCRIPTION.md @@ -14,6 +14,20 @@ entries more easily. This modules requires a secure context for the browser to work properly and therefore HTTPS support is required. +Vault Custodians can be configured in the general settings. These users +are automatically added to every newly created vault and can not be +removed from it. They keep access to the end-to-end encrypted vaults, +for example to recover the data in case an employee leaves the company. +Custodians receive read and share permissions by default; the owner can +additionally grant them write and delete permissions. + +The custodian protection only applies while a user is configured as a +custodian. Removing a user from the setting, or the user invalidating +their own keys, drops their access again. Because the encryption happens +in the browser, a custodian added to a vault created outside the browser +(e.g. by import or another module) only receives a usable key once a +user holding the master key re-shares or re-encrypts the vault. + The [vault-recovery](https://github.com/fkantelberg/vault-recovery) project focuses on disaster recovery in case of an incident to recover secrets from old database backups or old exports. diff --git a/vault/tests/__init__.py b/vault/tests/__init__.py index 59ea2f2ef1..2cc9ff45e0 100644 --- a/vault/tests/__init__.py +++ b/vault/tests/__init__.py @@ -3,6 +3,7 @@ from . import ( test_controller, + test_custodian, test_log, test_rights, test_user, diff --git a/vault/tests/test_custodian.py b/vault/tests/test_custodian.py new file mode 100644 index 0000000000..3eb09ec464 --- /dev/null +++ b/vault/tests/test_custodian.py @@ -0,0 +1,202 @@ +# © 2026 Nitrokey GmbH +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl). + +from odoo.exceptions import UserError +from odoo.tests import new_test_user + +from odoo.addons.base.tests.common import BaseCommon + + +class TestCustodian(BaseCommon): + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.custodian = cls._create_keyed_user("test-vault-custodian") + cls.owner = cls._create_keyed_user("test-vault-owner") + cls.env.company.vault_custodian_ids = [(6, 0, cls.custodian.ids)] + + @classmethod + def _create_keyed_user(cls, login): + """Create a user with vault keys so it can be used as a custodian.""" + user = new_test_user(cls.env, login=login) + cls.env["res.users.key"].create( + { + "user_id": user.id, + "public": "a public key", + "salt": "42", + "iv": "2424", + "iterations": 4000, + "private": "24", + "current": True, + } + ) + return user + + def _owner_right_vals(self): + return { + "user_id": self.env.uid, + "perm_create": True, + "perm_write": True, + "perm_delete": True, + "perm_share": True, + } + + def _create_vault(self, **vals): + return self.env["vault"].create({"name": "Vault", **vals}) + + def _create_vault_as_owner(self, **vals): + return self.env["vault"].with_user(self.owner).create({"name": "Vault", **vals}) + + def _custodian_right(self, vault, user=None): + user = user or self.custodian + return vault.right_ids.filtered(lambda r: r.user_id == user) + + # -- Seeding on vault creation ------------------------------------------ + + def test_custodian_added_on_create(self): + vault = self._create_vault() + right = self._custodian_right(vault) + self.assertEqual(right.user_id, self.custodian) + self.assertTrue(right.perm_share) + self.assertFalse(right.perm_create) + self.assertFalse(right.perm_write) + self.assertFalse(right.perm_delete) + + def test_multiple_custodians_added_on_create(self): + other = self._create_keyed_user("test-vault-custodian-2") + self.env.company.vault_custodian_ids = [(6, 0, (self.custodian + other).ids)] + vault = self._create_vault() + self.assertIn(self.custodian, vault.right_ids.user_id) + self.assertIn(other, vault.right_ids.user_id) + + def test_custodian_not_added_to_existing_vault(self): + # Only future vaults get the custodian; disabling here simulates a + # vault created before the custodian was configured + self.env.company.vault_custodian_ids = [(5, 0, 0)] + vault = self._create_vault() + self.assertFalse(self._custodian_right(vault)) + + def test_owner_as_custodian_no_duplicate(self): + # If the owner is also a custodian there must be no duplicated right + self.env.company.vault_custodian_ids = [(6, 0, self.env.user.ids)] + vault = self._create_vault() + self.assertEqual(len(vault.right_ids), 1) + + # -- Enforcement on create ---------------------------------------------- + + def test_custodian_removed_before_save_is_blocked(self): + # Removing the default custodian line before the first save (so unlink + # is never called) must still be prevented on create + with self.assertRaisesRegex(UserError, "must be shared"): + self._create_vault(right_ids=[(0, 0, self._owner_right_vals())]) + + def test_missing_one_of_multiple_custodians_blocked(self): + other = self._create_keyed_user("test-vault-custodian-2") + self.env.company.vault_custodian_ids = [(6, 0, (self.custodian + other).ids)] + with self.assertRaisesRegex(UserError, "must be shared"): + self._create_vault( + right_ids=[ + (0, 0, self._owner_right_vals()), + (0, 0, {"user_id": self.custodian.id, "perm_share": True}), + ] + ) + + def test_custodian_readd_without_share_is_blocked(self): + # Re-adding a custodian without the share permission must be blocked + with self.assertRaisesRegex(UserError, "must have the share permission"): + self._create_vault( + right_ids=[ + (0, 0, self._owner_right_vals()), + (0, 0, {"user_id": self.custodian.id, "perm_share": False}), + ] + ) + + # -- Protection of existing custodian rights ---------------------------- + + def test_custodian_cannot_be_removed(self): + vault = self._create_vault_as_owner() + right = self._custodian_right(vault) + with self.assertRaisesRegex(UserError, "can not be removed"): + right.with_user(self.owner).unlink() + + def test_custodian_cannot_be_removed_by_custodian(self): + # The custodian itself can not drop its own mandatory right either + vault = self._create_vault_as_owner() + right = self._custodian_right(vault) + with self.assertRaisesRegex(UserError, "can not be removed"): + right.with_user(self.custodian).unlink() + + def test_custodian_share_cannot_be_removed(self): + vault = self._create_vault_as_owner() + right = self._custodian_right(vault) + with self.assertRaisesRegex(UserError, "share permission"): + right.with_user(self.owner).perm_share = False + + def test_custodian_user_cannot_be_changed(self): + vault = self._create_vault_as_owner() + right = self._custodian_right(vault) + other = self._create_keyed_user("test-vault-other") + with self.assertRaisesRegex(UserError, "user of a custodian"): + right.with_user(self.owner).user_id = other + + def test_custodian_extra_permissions_can_be_granted(self): + vault = self._create_vault() + right = self._custodian_right(vault) + right.write({"perm_write": True, "perm_delete": True}) + self.assertTrue(right.perm_write) + self.assertTrue(right.perm_delete) + + def test_custodian_removable_as_superuser(self): + vault = self._create_vault() + right = self._custodian_right(vault) + right.sudo().unlink() + self.assertFalse(right.exists()) + + def test_vault_with_custodian_can_be_deleted(self): + vault = self._create_vault_as_owner() + right = self._custodian_right(vault) + vault.with_user(self.owner).unlink() + self.assertFalse(vault.exists()) + self.assertFalse(right.exists()) + + def test_custodian_removable_after_unconfigured(self): + # Protection is evaluated live: once the user is no longer configured + # as a custodian the right becomes a normal removable share + vault = self._create_vault() + right = self._custodian_right(vault) + self.env.company.vault_custodian_ids = [(5, 0, 0)] + right.unlink() + self.assertFalse(right.exists()) + + # -- Configuration validation ------------------------------------------- + + def test_keyless_user_cannot_be_custodian(self): + keyless = new_test_user(self.env, login="test-vault-keyless") + with self.assertRaisesRegex(UserError, "no vault keys"): + self.env.company.vault_custodian_ids = [(4, keyless.id)] + + def test_custodian_key_check_ignores_record_rules(self): + # A user with keys can be set as custodian even when the acting user + # can not read the custodian's keys due to record rules + admin = new_test_user( + self.env, login="test-vault-admin", groups="base.group_system" + ) + company = self.env.company.with_user(admin) + company.vault_custodian_ids = [(6, 0, self.custodian.ids)] + self.assertEqual(company.vault_custodian_ids, self.custodian) + + # -- has_vault_key helper ----------------------------------------------- + + def test_has_vault_key_compute(self): + keyless = new_test_user(self.env, login="test-vault-keyless") + self.assertTrue(self.custodian.has_vault_key) + self.assertFalse(keyless.has_vault_key) + + def test_has_vault_key_search(self): + keyless = new_test_user(self.env, login="test-vault-keyless") + with_keys = self.env["res.users"].search([("has_vault_key", "=", True)]) + without_keys = self.env["res.users"].search([("has_vault_key", "=", False)]) + self.assertIn(self.custodian, with_keys) + self.assertNotIn(self.custodian, without_keys) + self.assertIn(keyless, without_keys) + self.assertNotIn(keyless, with_keys) diff --git a/vault/views/res_config_settings_views.xml b/vault/views/res_config_settings_views.xml index df25fcdea6..8ab2f2b0e1 100644 --- a/vault/views/res_config_settings_views.xml +++ b/vault/views/res_config_settings_views.xml @@ -26,6 +26,17 @@ > + + +