diff --git a/src/pass_collaborate/model/auth.py b/src/pass_collaborate/model/auth.py index ce40383..fd19df0 100644 --- a/src/pass_collaborate/model/auth.py +++ b/src/pass_collaborate/model/auth.py @@ -180,16 +180,26 @@ def _check_auth_file(self, filename: str) -> Path: return config_file def _load_gpg_id_files(self) -> None: - """Load the data of the gpg-id files that is not already in the access store.""" + """Import the gpg-id files that are not yet in the access store. + + Merging the keys of an already imported gpg-id file would undo any + revocation done on the access store, so they are only warned about. + """ for gpg_id in self.store_dir.rglob(".gpg-id"): access = self.gpg_id_access_key(gpg_id) - try: - existent_keys = self.access_keys(access) - for key in gpg_id.read_text().splitlines(): - if key not in existent_keys: - self.access[access].append(key) - except KeyError: - self.access[access] = gpg_id.read_text().splitlines() + keys = gpg_id.read_text().splitlines() + + if access not in self.access: + self.access[access] = keys + continue + + unknown = [key for key in keys if key not in self.access_keys(access)] + if unknown: + log.warning( + f"{gpg_id} contains keys that are not in the access store, " + f"they will be removed on the next reencryption: " + f"{', '.join(unknown)}" + ) def save(self) -> None: """Save the contents of the authentication store.""" @@ -466,6 +476,10 @@ def _revoke_access(self, identifier: Identifier, access: List[str]) -> List[str] # We may pass here when we remove the access of a user to a group, # therefore the access to the directory doesn't change as it's # binded to the group + if isinstance(revoke, User): + # An access entry can also be the key or the email of the user. + return [entry for entry in access if not revoke.match(entry)] + with suppress(ValueError): access.remove(revoke.name) diff --git a/src/pass_collaborate/model/pass_.py b/src/pass_collaborate/model/pass_.py index a063b68..3822924 100644 --- a/src/pass_collaborate/model/pass_.py +++ b/src/pass_collaborate/model/pass_.py @@ -239,7 +239,7 @@ def update_gpg_id_file( gpg_id_file = gpg_id log.info(f"Updating the keys stored in {gpg_id}") keys.extend(self.allowed_keys(gpg_id_file)) - gpg_id_file.write_text("\n".join(set(keys)) + "\n", encoding="utf-8") + gpg_id_file.write_text("\n".join(sorted(set(keys))) + "\n", encoding="utf-8") def can_decrypt(self, path: Path) -> bool: """Test if the user can decrypt a file. diff --git a/tests/e2e/test_group.py b/tests/e2e/test_group.py index ede66ca..b51c046 100644 --- a/tests/e2e/test_group.py +++ b/tests/e2e/test_group.py @@ -436,3 +436,22 @@ def test_group_revoke_is_idempotent( assert result.exit_code == 0 pass_.auth.reload() assert pass_.auth.access == {".gpg-id": [admin.key]} + + +def test_revoke_a_user_that_is_in_the_access_by_key( + pass_: "PassStore", pass_dev: "PassStore", developer: "User" +) -> None: + """ + Given: A configured environment and a user authorized by their gpg key + When: Revoking their access + Then: The access is revoked and it's not restored from the gpg-id file + """ + pass_.auth.add_user(name=developer.name, email=developer.email, key=developer.key) + pass_.change_access("web", [developer.key]) + assert pass_dev.can_decrypt(pass_.path("web/production")) + + pass_.change_access("web", remove_identifiers=[developer.email]) # act + + pass_.auth.reload() + assert developer.key not in pass_.auth.access["web/.gpg-id"] + assert not pass_dev.can_decrypt(pass_.path("web/production")) diff --git a/tests/unit/adapters/test_authstore.py b/tests/unit/adapters/test_authstore.py index a79e12f..9ea7dcf 100644 --- a/tests/unit/adapters/test_authstore.py +++ b/tests/unit/adapters/test_authstore.py @@ -10,6 +10,8 @@ from ...factories import GroupFactory, UserFactory if TYPE_CHECKING: + from _pytest.logging import LogCaptureFixture + from pass_collaborate.model.auth import AuthStore from pass_collaborate.model.key import GPGKey @@ -175,16 +177,18 @@ def test_user_can_have_accents_on_name() -> None: assert user == result -def test_auth_loads_gpg_id_even_if_entry_exists( - auth: "AuthStore", admin: "User" +def test_auth_doesnt_load_gpg_id_if_entry_exists( + auth: "AuthStore", admin: "User", caplog: "LogCaptureFixture" ) -> None: """ - Given: an auth store with a key in the access property - Then: The missing keys are loaded + Given: an auth store with an already imported gpg-id file + When: the gpg-id file has keys that the access store doesn't have + Then: The keys are not loaded, so a revocation is not undone, but warned about """ auth.access[".gpg-id"] = [] auth.save() auth.reload() # act - assert auth.access[".gpg-id"] == [admin.key] + assert auth.access[".gpg-id"] == [] + assert admin.key in caplog.records[0].message