updating groups/machines/users keys now also update vars secrets

pkgs/clan-cli/clan_cli/secrets/groups.py

@@ -20,7 +20,6 @@ from .folders import (
     sops_secrets_folder,
     sops_users_folder,
 )
-from .sops import update_keys
 from .types import (
     VALID_USER_NAME,
     group_name_type,
@@ -97,15 +96,10 @@ def list_directory(directory: Path) -> str:
 
 
 def update_group_keys(flake_dir: Path, group: str) -> list[Path]:
-    updated_paths = []
+    def filter_group_secrets(secret: Path) -> bool:
-    for secret_ in secrets.list_secrets(flake_dir):
+        return (secret / "groups" / group).is_symlink()
-        secret = sops_secrets_folder(flake_dir) / secret_
+
-        if (secret / "groups" / group).is_symlink():
+    return secrets.update_secrets(flake_dir, filter_secrets=filter_group_secrets)
-            updated_paths += update_keys(
-                secret,
-                secrets.collect_keys_for_path(secret),
-            )
-    return updated_paths
 
 
 def add_member(
@@ -209,6 +203,21 @@ def add_secret(flake_dir: Path, group: str, name: str) -> None:
     )
 
 
+def get_groups(
+    flake_dir: Path,
+    type_name: str,
+    name: str,
+) -> list[Path]:
+    groups_dir = sops_groups_folder(flake_dir)
+
+    groups = []
+    if groups_dir.exists():
+        for group in groups_dir.iterdir():
+            if group.is_dir() and (group / type_name / name).exists():
+                groups.append(group)
+    return groups
+
+
 def add_secret_command(args: argparse.Namespace) -> None:
     add_secret(args.flake.path, args.group, args.secret)
 

pkgs/clan-cli/clan_cli/secrets/machines.py

@@ -17,24 +17,29 @@ from .folders import (
     sops_machines_folder,
     sops_secrets_folder,
 )
+from .groups import get_groups
 from .secrets import update_secrets
 from .sops import read_key, write_key
 from .types import public_or_private_age_key_type, secret_name_type
 
 
-def add_machine(flake_dir: Path, machine: str, pubkey: str, force: bool) -> None:
+def add_machine(flake_dir: Path, name: str, pubkey: str, force: bool) -> None:
-    machine_path = sops_machines_folder(flake_dir) / machine
+    machine_path = sops_machines_folder(flake_dir) / name
     write_key(machine_path, pubkey, sops.KeyType.AGE, overwrite=force)
     paths = [machine_path]
 
+    groups = get_groups(flake_dir, "machines", name)
+
     def filter_machine_secrets(secret: Path) -> bool:
-        return (secret / "machines" / machine).exists()
+        if (secret / "machines" / name).exists():
+            return True
+        return any(secret.joinpath("groups", group.name).exists() for group in groups)
 
     paths.extend(update_secrets(flake_dir, filter_secrets=filter_machine_secrets))
     commit_files(
         paths,
         flake_dir,
-        f"Add machine {machine} to secrets",
+        f"Add machine {name} to secrets",
     )
 
 

pkgs/clan-cli/clan_cli/secrets/secrets.py

@@ -1,4 +1,5 @@
 import argparse
+import functools
 import getpass
 import logging
 import os
@@ -39,18 +40,52 @@ from .types import VALID_SECRET_NAME, secret_name_type
 log = logging.getLogger(__name__)
 
 
+def list_generators_secrets(generators_path: Path) -> list[Path]:
+    for generator_path in generators_path.iterdir():
+        if not generator_path.is_dir():
+            continue
+
+        def validate(generator_path: Path, name: str) -> bool:
+            return has_secret(generator_path / name)
+
+        paths = []
+        for obj in list_objects(
+            generator_path, functools.partial(validate, generator_path)
+        ):
+            paths.append(generator_path / obj)
+        return paths
+    return []
+
+
+def list_vars_secrets(flake_dir: Path) -> list[Path]:
+    secret_paths = []
+    shared_dir = flake_dir / "vars" / "shared"
+    if shared_dir.is_dir():
+        secret_paths.extend(list_generators_secrets(shared_dir))
+
+    machines_dir = flake_dir / "vars" / "per-machine"
+    if machines_dir.is_dir():
+        for machine_dir in machines_dir.iterdir():
+            if not machine_dir.is_dir():
+                continue
+            secret_paths.extend(list_generators_secrets(machine_dir))
+    return secret_paths
+
+
 def update_secrets(
     flake_dir: Path, filter_secrets: Callable[[Path], bool] = lambda _: True
 ) -> list[Path]:
     changed_files = []
-    for name in list_secrets(flake_dir):
+    secret_paths = [sops_secrets_folder(flake_dir) / s for s in list_secrets(flake_dir)]
-        secret_path = sops_secrets_folder(flake_dir) / name
+    secret_paths.extend(list_vars_secrets(flake_dir))
-        if not filter_secrets(secret_path):
+
+    for path in secret_paths:
+        if not filter_secrets(path):
             continue
         changed_files.extend(
             update_keys(
-                secret_path,
+                path,
-                collect_keys_for_path(secret_path),
+                collect_keys_for_path(path),
             )
         )
     return changed_files
@@ -222,7 +257,7 @@ def allow_member(
 ) -> list[Path]:
     source = source_folder / name
     if not source.exists():
-        msg = f"Cannot encrypt {group_folder.parent.name} for '{name}' group. '{name}' group does not exist in {source_folder}: "
+        msg = f"Cannot encrypt {group_folder.parent.name} for '{name}'. '{name}' does not exist in {source_folder}: "
         msg += list_directory(source_folder)
         raise ClanError(msg)
     group_folder.mkdir(parents=True, exist_ok=True)

pkgs/clan-cli/clan_cli/secrets/users.py

@@ -8,7 +8,13 @@ from clan_cli.errors import ClanError
 from clan_cli.git import commit_files
 
 from . import secrets, sops
-from .folders import list_objects, remove_object, sops_secrets_folder, sops_users_folder
+from .folders import (
+    list_objects,
+    remove_object,
+    sops_secrets_folder,
+    sops_users_folder,
+)
+from .groups import get_groups
 from .secrets import update_secrets
 from .sops import read_key, write_key
 from .types import (
@@ -28,11 +34,16 @@ def add_user(
 ) -> None:
     path = sops_users_folder(flake_dir) / name
 
+    groups = get_groups(flake_dir, "users", name)
+
     def filter_user_secrets(secret: Path) -> bool:
-        return secret.joinpath("users", name).exists()
+        if secret.joinpath("users", name).exists():
+            return True
+        return any(secret.joinpath("groups", group.name).exists() for group in groups)
 
     write_key(path, key, key_type, overwrite=force)
     paths = [path]
+
     paths.extend(update_secrets(flake_dir, filter_secrets=filter_user_secrets))
     commit_files(
         paths,

pkgs/clan-cli/tests/test_vars.py

@@ -2,6 +2,7 @@ import json
 import logging
 import shutil
 from pathlib import Path
+from typing import TYPE_CHECKING
 
 import pytest
 from age_keys import SopsSetup
@@ -20,6 +21,9 @@ from clan_cli.vars.set import set_var
 from fixtures_flakes import ClanFlake
 from helpers import cli
 
+if TYPE_CHECKING:
+    from age_keys import KeyPair
+
 
 def test_dependencies_as_files(temp_dir: Path) -> None:
     from clan_cli.vars.generate import dependencies_as_dir
@@ -218,6 +222,7 @@ def test_generate_secret_var_sops_with_default_group(
     monkeypatch: pytest.MonkeyPatch,
     flake: ClanFlake,
     sops_setup: SopsSetup,
+    age_keys: list["KeyPair"],
 ) -> None:
     config = flake.machines["my_machine"]
     config["nixpkgs"]["hostPlatform"] = "x86_64-linux"
@@ -242,10 +247,9 @@ def test_generate_secret_var_sops_with_default_group(
     )
     assert sops_store.exists(Generator("my_generator"), "my_secret")
     assert sops_store.get(Generator("my_generator"), "my_secret").decode() == "hello\n"
-    # add another user and check if secret gets re-encrypted
-    from clan_cli.secrets.sops import generate_private_key
 
-    _, pubkey_uschi = generate_private_key()
+    # add another user to the group and check if secret gets re-encrypted
+    pubkey_user2 = age_keys[1]
     cli.run(
         [
             "secrets",
@@ -253,19 +257,34 @@ def test_generate_secret_var_sops_with_default_group(
             "add",
             "--flake",
             str(flake.path),
-            "uschi",
+            "user2",
-            pubkey_uschi,
+            pubkey_user2.pubkey,
         ]
     )
-    cli.run(["secrets", "groups", "add-user", "my_group", "uschi"])
+    cli.run(["secrets", "groups", "add-user", "my_group", "user2"])
-    with pytest.raises(ClanError):
-        cli.run(["vars", "generate", "--flake", str(flake.path), "my_machine"])
-    # apply fix
-    cli.run(["vars", "fix", "--flake", str(flake.path), "my_machine"])
     # check if new user can access the secret
-    monkeypatch.setenv("USER", "uschi")
+    monkeypatch.setenv("USER", "user2")
     assert sops_store.user_has_access(
-        "uschi", Generator("my_generator", share=False), "my_secret"
+        "user2", Generator("my_generator", share=False), "my_secret"
+    )
+
+    # Rotate key of a user
+    pubkey_user3 = age_keys[2]
+    cli.run(
+        [
+            "secrets",
+            "users",
+            "add",
+            "--flake",
+            str(flake.path),
+            "--force",
+            "user2",
+            pubkey_user3.pubkey,
+        ]
+    )
+    monkeypatch.setenv("USER", "user2")
+    assert sops_store.user_has_access(
+        "user2", Generator("my_generator", share=False), "my_secret"
     )