yadunut/clan-core
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

vars: implement dependencies

Browse Source
This commit is contained in:
DavHau
2024-07-17 16:42:16 +07:00
parent cfda89ef01
commit 566c1403c0
4 changed files with 183 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

124
pkgs/clan-cli/clan_cli/vars/generate.py
View File

@@ -5,6 +5,7 @@ import os
import subprocess
import sys
from collections.abc import Callable
from graphlib import TopologicalSorter
from pathlib import Path
from tempfile import TemporaryDirectory
@@ -37,7 +38,7 @@ def read_multiline_input(prompt: str = "Finish with Ctrl-D") -> str:
return proc.stdout
def bubblewrap_cmd(generator: str, generator_dir: Path) -> list[str]:
def bubblewrap_cmd(generator: str, generator_dir: Path, dep_tmpdir: Path) -> list[str]:
# fmt: off
return run_cmd(
[
@@ -50,6 +51,7 @@ def bubblewrap_cmd(generator: str, generator_dir: Path) -> list[str]:
"--tmpfs", "/usr/lib/systemd",
"--dev", "/dev",
"--bind", str(generator_dir), str(generator_dir),
"--ro-bind", str(dep_tmpdir), str(dep_tmpdir),
"--unshare-all",
"--unshare-user",
"--uid", "1000",
@@ -60,16 +62,58 @@ def bubblewrap_cmd(generator: str, generator_dir: Path) -> list[str]:
# fmt: on
# TODO: implement caching to not decrypt the same secret multiple times
def decrypt_dependencies(
machine: Machine,
generator_name: str,
secret_vars_store: SecretStoreBase,
public_vars_store: FactStoreBase,
) -> dict[str, dict[str, bytes]]:
generator = machine.vars_generators[generator_name]
dependencies = set(generator["dependencies"])
decrypted_dependencies = {}
for dep_generator in dependencies:
decrypted_dependencies[dep_generator] = {}
dep_files = machine.vars_generators[dep_generator]["files"]
for file_name, file in dep_files.items():
if file["secret"]:
decrypted_dependencies[dep_generator][file_name] = (
secret_vars_store.get(dep_generator, file_name)
)
else:
decrypted_dependencies[dep_generator][file_name] = (
public_vars_store.get(dep_generator, file_name)
)
return decrypted_dependencies
# decrypt dependencies and return temporary file tree
def dependencies_as_dir(
decrypted_dependencies: dict[str, dict[str, bytes]],
tmpdir: Path,
) -> Path:
for dep_generator, files in decrypted_dependencies.items():
dep_generator_dir = tmpdir / dep_generator
dep_generator_dir.mkdir()
dep_generator_dir.chmod(0o700)
for file_name, file in files.items():
file_path = dep_generator_dir / file_name
file_path.touch()
file_path.chmod(0o600)
file_path.write_bytes(file)
return tmpdir
def execute_generator(
machine: Machine,
generator_name: str,
regenerate: bool,
secret_vars_store: SecretStoreBase,
public_vars_store: FactStoreBase,
tmpdir: Path,
dep_tmpdir: Path,
prompt: Callable[[str], str],
) -> bool:
generator_dir = tmpdir / generator_name
generator_dir = dep_tmpdir / generator_name
# check if all secrets exist and generate them if at least one is missing
needs_regeneration = not check_secrets(machine, generator_name=generator_name)
log.debug(f"{generator_name} needs_regeneration: {needs_regeneration}")
@@ -79,22 +123,30 @@ def execute_generator(
msg = f"flake is not a Path: {machine.flake}"
msg += "fact/secret generation is only supported for local flakes"
env = os.environ.copy()
generator_dir.mkdir(parents=True)
env["out"] = str(generator_dir)
# compatibility for old outputs.nix users
generator = machine.vars_generators[generator_name]["finalScript"]
# if machine.vars_data[generator_name]["generator"]["prompt"]:
# prompt_value = prompt(machine.vars_data[generator_name]["generator"]["prompt"])
# env["prompt_value"] = prompt_value
if sys.platform == "linux":
cmd = bubblewrap_cmd(generator, generator_dir)
else:
cmd = ["bash", "-c", generator]
run(
cmd,
env=env,
# build temporary file tree of dependencies
decrypted_dependencies = decrypt_dependencies(
machine, generator_name, secret_vars_store, public_vars_store
)
env = os.environ.copy()
generator_dir.mkdir(parents=True)
env["out"] = str(generator_dir)
with TemporaryDirectory() as tmp:
dep_tmpdir = dependencies_as_dir(decrypted_dependencies, Path(tmp))
env["in"] = str(dep_tmpdir)
if sys.platform == "linux":
cmd = bubblewrap_cmd(generator, generator_dir, dep_tmpdir=dep_tmpdir)
else:
cmd = ["bash", "-c", generator]
run(
cmd,
env=env,
)
files_to_commit = []
# store secrets
files = machine.vars_generators[generator_name]["files"]
@@ -129,6 +181,17 @@ def prompt_func(text: str) -> str:
return read_multiline_input()
def _get_subgraph(graph: dict[str, set], vertex: str) -> dict[str, set]:
visited = set()
queue = [vertex]
while queue:
vertex = queue.pop(0)
if vertex not in visited:
visited.add(vertex)
queue.extend(graph[vertex] - visited)
return {k: v for k, v in graph.items() if k in visited}
def _generate_vars_for_machine(
machine: Machine,
generator_name: str | None,
@@ -152,21 +215,40 @@ def _generate_vars_for_machine(
f"Could not find generator with name: {generator_name}. The following generators are available: {generators}"
)
if generator_name:
machine_generator_facts = {
generator_name: machine.vars_generators[generator_name]
}
else:
machine_generator_facts = machine.vars_generators
# if generator_name:
# machine_generator_facts = {
# generator_name: machine.vars_generators[generator_name]
# }
# else:
# machine_generator_facts = machine.vars_generators
for generator_name in machine_generator_facts:
graph = {
gen_name: set(generator["dependencies"])
for gen_name, generator in machine.vars_generators.items()
}
# extract sub-graph if specific generator selected
if generator_name:
graph = _get_subgraph(graph, generator_name)
# check if all dependencies actually exist
for gen_name, dependencies in graph.items():
for dep in dependencies:
if dep not in graph:
raise ClanError(
f"Generator {gen_name} has a dependency on {dep}, which does not exist"
)
# process generators in topological order
sorter = TopologicalSorter(graph)
for generator_name in sorter.static_order():
machine_updated |= execute_generator(
machine=machine,
generator_name=generator_name,
regenerate=regenerate,
secret_vars_store=secret_vars_store,
public_vars_store=public_vars_store,
tmpdir=local_temp,
dep_tmpdir=local_temp,
prompt=prompt,
)
if machine_updated: