Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 22 additions & 8 deletions src/pass_collaborate/model/auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand Down Expand Up @@ -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)

Expand Down
2 changes: 1 addition & 1 deletion src/pass_collaborate/model/pass_.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
19 changes: 19 additions & 0 deletions tests/e2e/test_group.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"))
14 changes: 9 additions & 5 deletions tests/unit/adapters/test_authstore.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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