tasks.py

import os
from pathlib import Path
from typing import Any, Union
import subprocess
import getpass
from tempfile import TemporaryDirectory
import json
import warnings
import platform
from contextlib import contextmanager, nullcontext

# REF; https://www.pyinvoke.org/
from invoke import task

# REF; https://github.com/numtide/deploykit/
# REF; https://github.com/takluyver/pyxdg/
from xdg.BaseDirectory import save_cache_path


INVOKED_PATH = Path.cwd()

PROJECT_DIR = Path(__file__).parent.resolve()
os.chdir(PROJECT_DIR)

CACHE_DIR = save_cache_path("proesmans")

FLAKE = PROJECT_DIR / "flake"
DOCS = PROJECT_DIR / "documentation"
DEV_KEY = FLAKE / "development.age"

# If the target host URL contains any of these values, assume a local/fast connection between build- and target host
LOCAL_TARGETS_MARKER = ["localhost", "127.0.0.1", "192.168.", ".internal.proesmans.eu"]

# Generate and store a new disk encryption key using;
# tr -dc '[:alnum:]' </dev/urandom | head -c64
#
# WARN; Path hardcoded in DISKO configuration !
REMOTE_LUKS_SECRET_PATH = "/tmp/deployment-disk.key"


def alert_finish():
    # Riiiing my bell ! Ring my bell ! TINGELINGELING
    print("\a")


def ask_user_input(message: str) -> bool:
    user_reply = input(f"{message} [y/N]: ")
    return user_reply in ["yes", "y"]


def decryptor_encrypted_filename_default() -> str:
    return "keys.encrypted.yaml"


def decryptor_name_default(hostname: str) -> str:
    return f"{hostname}_decrypter"


def find_string_in_file(file_path, needle):
    try:
        with open(file_path, "r") as file:
            for line in file:
                if needle in line:
                    return True
        return False
    except FileNotFoundError:
        print(f"find_string_in_file: The file at {file_path} was not found.")
        raise
    except Exception as e:
        print(f"find_string_in_file: An error occurred: {e}")
        raise


def private_opener(path: str, flags: int) -> Union[str, int]:
    return os.open(path, flags, 0o400)


def dev_key_decrypt() -> str:
    assert DEV_KEY.exists(), """
        The encrypted development key is not found next to the tasks.py file!
    """

    warnings.warn("Decrypting the development key for usage!")
    age_key = subprocess.run(
        ["rage", "--decrypt", DEV_KEY.as_posix()],
        text=True,  # stdin/stdout are opened in text mode
        check=True,  # Throw exception if command fails
        capture_output=True,  # Redirect stdout/stderr
    ).stdout.strip()

    assert age_key, """
        Expected to decrypt an AGE private key, but decrypted an empty string. Something went unexpectedly wrong!
    """

    return age_key


def get_verified_password() -> str:
    while True:
        first = getpass.getpass("Enter password: ")
        second = getpass.getpass("Confirm password: ")
        if first == second:
            return first
        else:
            print("Passwords do not match. Try again.")


@contextmanager
def pipe_with_data(data: bytes):
    # WARN; read_descriptor is an INTEGER !
    read_descriptor, write_descriptor = os.pipe()
    try:
        os.write(write_descriptor, data)
        os.close(write_descriptor)
        yield read_descriptor
    finally:
        os.close(read_descriptor)


@task
# USAGE: invoke check all|<hostName>
def check(c: Any, hostName: str) -> None:
    """
    Evaluate and build all outputs from the flake common schema, including all attribute sets from the output 'checks'.
    This command does not stop executing after encountering an error, and will run until all tasks have ended.
    """
    if "all" == hostName:
        # NOTE; --skip-cached skips realized outputs already present in binary caches!
        c.run(f"nix-fast-build --no-link --flake {FLAKE}#checks")
    else:
        c.run(
            f"nix-fast-build --no-link --flake {FLAKE}#nixosConfigurations.{hostName}.config.system.build.toplevel"
        )
    alert_finish()


@task
def ci(c: Any) -> None:
    """
    Similar to task 'check', but also builds the no-system jobs!
    """
    system = subprocess.run(
        ["nix", "eval", "--raw", "--impure", "--expr", "builtins.currentSystem"],
        text=True,  # stdin/stdout are opened in text mode
        check=True,
        capture_output=True,
    ).stdout.strip()

    # NOTE; --skip-cached skips realized outputs already present in binary caches!
    c.run(
        f'nix-fast-build --no-nom --skip-cached --no-link --flake "{FLAKE}#hydraJobs.{system}"'
    )

    if "x86_64-linux" == system:
        c.run(
            f"nix-fast-build --no-nom --skip-cached --no-link --flake '{FLAKE}#hydraJobs.no-system'"
        )
    alert_finish()


@task
# USAGE: invoke sops-files-update
def sops_files_update(c: Any) -> None:
    """
    Update all sops files according to .sops.yaml rules
    """
    environment = os.environ.copy()
    environment.pop("SOPS_AGE_KEY_FILE", None)
    environment["SOPS_AGE_KEY"] = dev_key_decrypt()

    subprocess.run(
        """
        find . -type f \\( -iname '*.encrypted.yaml' -o -iname '*.encrypted.json' \\) -print0 | \
        xargs -0 -n1 sops updatekeys --yes
        """,
        cwd=FLAKE,
        env=environment,
        shell=True,
        check=True,
    )


@task
# USAGE; invoke deploy development root@10.1.7.100 [-k "development_decrypter"] [-p]
def deploy(
    c: Any,
    hostname: str,
    ssh_connection_string: str,
    key: str = None,
    password_request: bool = False,
) -> None:
    """
    Decrypts the secret used for sops-nix, deploys the machine, upload the secret to the host filesystem.
    """

    host_configuration_dir = FLAKE / "nixosConfigurations" / hostname
    encrypted_file = host_configuration_dir / decryptor_encrypted_filename_default()
    # Development host is x86, but these systems are aarch64!
    mismatched_architecture = hostname in ["freddy"]

    assert host_configuration_dir.is_dir(), f"""
        There is no configuration folder found for host {hostname}.
        Create a nixos configuration at path `{host_configuration_dir.as_posix()}` first!
    """

    if not encrypted_file.is_file():
        raise FileNotFoundError(
            f"No decrypter keys file found. Create a decrypter key for host {hostname} first!"
        )

    host_attr_path = (
        f"{FLAKE}#nixosConfigurations.{hostname}.config.system.build.toplevel"
    )

    print(f"Checking if host {hostname} builds..")
    subprocess.run(["nix", "build", host_attr_path, "--no-link"], check=True)

    if not ask_user_input(
        f"Install configuration {hostname} on {ssh_connection_string}?"
    ):
        return

    if not key:
        key = decryptor_name_default(hostname)
        warnings.warn(f"Defaulting to SSH hostkey secret name: {key}")

    # Request user to provide disk encryption password
    password_generator = None
    if password_request:
        print("Please provide a DISK encryption secret")
        # Generate and store a new key using;
        # tr -dc '[:alnum:]' </dev/urandom | head -c64
        password_generator = pipe_with_data(get_verified_password().encode("UTF-8"))

    environment = os.environ.copy()
    environment.pop("SOPS_AGE_KEY_FILE", None)
    environment["SOPS_AGE_KEY"] = dev_key_decrypt()

    print(f"Decrypting AGE identity from {encrypted_file}:{key}..")
    age_key = subprocess.run(
        [
            "sops",
            "decrypt",
            "--extract",
            json.dumps([key]),
            encrypted_file.as_posix(),
        ],
        env=environment,
        text=True,  # stdin/stdout are opened in text mode
        check=True,
        capture_output=True,
    ).stdout.strip()

    if not age_key:
        raise RuntimeError("Decrypted AGE private key is empty")

    with TemporaryDirectory() as deploy_directory:
        # Prepare filepath and secure file access to store sensitive key material
        deploy_directory = Path(deploy_directory)
        deploy_directory.mkdir(parents=True, exist_ok=True)
        deploy_directory.chmod(0o755)
        decrypter_file_path = deploy_directory / "etc" / "secrets" / "decrypter.age"
        decrypter_file_path.parent.mkdir(parents=True, exist_ok=True)

        with open(decrypter_file_path, "wt", opener=private_opener) as file_handle:
            file_handle.write(age_key)

        deploy_flags = []
        deploy_flags.append("--debug")
        # "--no-substitute-on-destination",
        # "--stop-after-disko", # DEBUG
        # "--no-reboot",
        # NOTE; Flakes can give hints to the nix CLI to change runtime behaviours, like adding a binary cache for
        # operations on that flake execution only.
        # These options are encoded inside the 'nixConfig' output attribute of the flake-schema.
        # REF; https://nixos.org/manual/nix/stable/command-ref/new-cli/nix3-flake.html#flake-format
        #
        # "--option accept-flake-config true",

        if mismatched_architecture:
            deploy_flags.append("--build-on")
            deploy_flags.append("remote")

        # NOTE; The (nixos-anywhere) default is to let the target pull packages from the caches first, and if they not exist there
        # the current (buildhost) host will push the packages.
        # There are more situations where uploading from current host first is desired, as opposed to downloading from
        # the internet caches!
        if any(x in ssh_connection_string for x in LOCAL_TARGETS_MARKER):
            # Since we have a populated nix store, and this is a local install; do not let the target pull from
            # the external nix caches.
            deploy_flags.append("--no-substitute-on-destination")

        with (
            password_generator or nullcontext()
        ) as password_descriptor:  # ->N (integer)
            # ERROR; Cannot use sops --exec-file because we need to pass a full file structure to nixos-anywhere
            subprocess.run(
                [
                    "nixos-anywhere",
                    "--extra-files",
                    deploy_directory,
                    "--flake",
                    f"{FLAKE}#{hostname}",
                    *deploy_flags,
                    *(
                        [
                            "--disk-encryption-keys",
                            REMOTE_LUKS_SECRET_PATH,
                            f"/proc/self/fd/{password_descriptor}",
                        ]
                        if password_request
                        else []
                    ),
                    ssh_connection_string,
                ],
                env=environment,
                pass_fds=(password_descriptor,) if password_request else (),
                check=True,
            )

        alert_finish()


@task
# USAGE; invoke filesystem-rebuild development
def filesystem_rebuild(c: Any, flake_attr: str) -> None:
    """
    Builds the disko format script, pushes it to the destination host and executes the script.
    This will attempt to realise the (presumably) changed configuration. This is only really useful when
    new ZFS datasets were added, or empty disk space is now taken in with new partition(s).

    This operation should happen **before** a nixos-rebuild applies the new system configuration!
    Probably best to restart first before updating the system build!
    """

    # NOTE; The format script will leave existing disks and partitions that are not defined within
    # the configuration intact.
    # NOTE; The format script will attempt to reapply partition attributes if they do not match
    # with the configuration.
    format_attr_path = (
        f"{FLAKE}#nixosConfigurations.{flake_attr}.config.system.build.formatScript"
    )

    print(f"Checking if format script builds for {flake_attr}..")
    format_script = subprocess.run(
        [
            "nix",
            "build",
            format_attr_path,
            "--no-link",
            "--no-eval-cache",
            "--print-out-paths",
        ],
        check=True,
        text=True,
        capture_output=True,
    ).stdout.strip()

    print(f"Evaluating machine facts to find {flake_attr}..")
    text_machines = subprocess.run(
        [
            "nix",
            "eval",
            "--json",
            f"{FLAKE}#facts",
            "--apply",
            # ERROR; The evalModule system asserts when accessing a config value for unset option.
            # ERROR; The to-JSON export function asserts when it encounters a function.
            # TODO; Use host-data before fallback to dns-name.
            'builtins.mapAttrs (_: v: "${v.hostName}.${v.domainName}")',
        ],
        check=True,
        text=True,
        capture_output=True,
    ).stdout.strip()

    machines = json.loads(text_machines)
    ssh_connection_string = next(
        (
            host_name
            for moniker, host_name in machines.items()
            if flake_attr == moniker or host_name.startswith(flake_attr)
        ),
        None,
    )

    if not ssh_connection_string:
        raise LookupError(
            "No ssh moniker found for host attribute. Set `proesmans.facts.hostName` in nixos config"
        )

    if not ask_user_input(
        f"Update filesystems for {flake_attr} on {ssh_connection_string}?"
    ):
        return

    subprocess.run(
        ["nix", "copy", "--to", f"ssh://{ssh_connection_string}", format_script],
        check=True,
    )

    subprocess.run(["ssh", ssh_connection_string, f"sudo {format_script}"], check=True)

    alert_finish()


@task
# USAGE; invoke dev-rebuild
def dev_rebuild(c: Any, boot: bool = False) -> None:
    """
    Rebuild the current machine with the host configuration for "development"
    """
    this_hostname = platform.node()
    if this_hostname != "development":
        raise RuntimeError(
            f"Hostname is '{this_hostname}', not 'development'. Refusing to clobber configuration"
        )

    if boot:
        print("== WARNING ==")
        print(
            "Boot flag used. You must reboot the host manually after nixos-rebuild is done!"
        )

    subprocess.run(
        [
            "sudo",
            "nixos-rebuild",
            "--flake",
            f"{FLAKE}#development",
            "boot" if boot else "switch",
        ],
        check=True,
    )

    if boot:
        print("== WARNING ==")
        print(
            "Boot flag used. You must reboot the host manually after nixos-rebuild is done!"
        )

    alert_finish()


@task
# USAGE; invoke unlock freddy
def unlock(c: Any, flake_attr: str) -> None:
    """
    Open an interactive session into the pre-boot environment of the host to provide disk decryption password.
    """
    print(f"Evaluating machine facts to find {flake_attr}..")
    text_machines = subprocess.run(
        [
            "nix",
            "eval",
            "--json",
            f"{FLAKE}#facts",
            "--apply",
            # ERROR; The evalModule system asserts when accessing a config value for unset option.
            # ERROR; The to-JSON export function asserts when it encounters a function.
            # TODO; Use host-data before fallback to dns-name.
            'builtins.mapAttrs (_: v: "${v.hostName}.${v.domainName}")',
        ],
        check=True,
        text=True,
        capture_output=True,
    ).stdout.strip()

    machines = json.loads(text_machines)
    ssh_connection_string = next(
        (
            host_name
            for moniker, host_name in machines.items()
            if flake_attr == moniker or host_name.startswith(flake_attr)
        ),
        None,
    )

    assert ssh_connection_string, """
        There is no ssh moniker found for the provided hostname.
        Make sure the desired host returns the expected option host-facts.<moniker>.hostName using the nixos configuration options `proesmans.facts.hostName = "<TODO>";`
    """
    subprocess.run(
        [
            "ssh",
            # Disable storing host keys, the host key is dynamic
            "-o",
            "StrictHostKeyChecking=no",
            # Skip verifying against known host keys, the host key is dynamic
            "-o",
            "UserKnownHostsFile=/dev/null",
            # WARN; Hardcoded port in priviledged range
            "-p",
            "23",
            # WARN; Assumed root user during boot phase-1
            f"root@{ssh_connection_string}",
        ],
        check=True,
    )


@task
# USAGE; invoke rebuild development
def rebuild(c: Any, flake_attr: str, yes: bool = False, boot: bool = False) -> None:
    """
    Build a host configuration and activate it on the machine.
    """
    host_attr_path = (
        f"{FLAKE}#nixosConfigurations.{flake_attr}.config.system.build.toplevel"
    )

    if boot:
        print("== WARNING ==")
        print(
            "Boot flag used. You must reboot the host manually after nixos-rebuild is done!"
        )

    if not yes:
        print(f"== Checking if host {flake_attr} builds ==")
        subprocess.run(
            ["nix-fast-build", "--flake", host_attr_path, "--no-link"], check=True
        )

    print(f"== Evaluating machine facts to find {flake_attr} ==")
    text_machines = subprocess.run(
        [
            "nix",
            "eval",
            "--json",
            f"{FLAKE}#facts",
            "--apply",
            # ERROR; The evalModule system asserts when accessing a config value for unset option.
            # ERROR; The to-JSON export function asserts when it encounters a function.
            # TODO; Use host-data before fallback to dns-name.
            'builtins.mapAttrs (_: v: "${v.hostName}.${v.domainName}")',
        ],
        check=True,
        text=True,
        capture_output=True,
    ).stdout.strip()

    machines = json.loads(text_machines)
    ssh_connection_string = next(
        (
            host_name
            for moniker, host_name in machines.items()
            if flake_attr == moniker or host_name.startswith(flake_attr)
        ),
        None,
    )

    if not ssh_connection_string:
        raise LookupError(
            "No ssh moniker found for hostname. Set `proesmans.facts.hostName` in nixos config"
        )

    if not yes:
        ask = input(
            f"Update configuration {flake_attr} on {ssh_connection_string}? [y/N] "
        )
        if ask != "y":
            return

    additional_switches = []
    additional_switches.append("--sudo")

    if not any(x in ssh_connection_string for x in LOCAL_TARGETS_MARKER):
        # Download as much from online caches because the link between development- and target host is slow.
        additional_switches.append("--use-substitutes")

    subprocess.run(
        [
            "nixos-rebuild",
            "--flake",
            f"{FLAKE}#{flake_attr}",
            "--target-host",
            ssh_connection_string,
            *additional_switches,
            "boot" if boot else "switch",
        ],
        check=True,
    )

    if boot:
        print("== WARNING ==")
        print(
            "Boot flag used. You must reboot the host manually after nixos-rebuild is done!"
        )

    print("== Pinning host closure as garbage root (nix gcroot) ==")
    # The machine builds and is deployed succesfully, pinning should succeed IF we have the closure downloaded locally
    realised_path_exec = subprocess.run(
        ["nix", "path-info", host_attr_path], text=True, capture_output=True
    )
    if realised_path_exec != 0:
        # There is no local copy of the system closure, build it first (implies copy to local system if remote-builder is used)..
        subprocess.run(
            ["nix-fast-build", "--flake", host_attr_path, "--no-link"], check=True
        )
        # ..then retry capturing the path of the built closure
        realised_path_exec = subprocess.run(
            ["nix", "path-info", host_attr_path],
            check=True,  # Fails if there is still no local copy
            text=True,
            capture_output=True,
        )

    subprocess.run(
        [
            "nix-store",
            "--add-root",
            f"{CACHE_DIR}/{flake_attr}.pin",
            "--realise",
            realised_path_exec.stdout.strip(),
        ],
        check=True,
    )
    alert_finish()


@task
# USAGE; invoke secret-edit development [-f "secrets.encrypted.yaml"] [--binary]
def secret_edit(
    c: Any,
    hostname: str,
    file: str = "secrets.encrypted.yaml",
    # Force interpretation of the file as "binary" type
    binary: bool = False,
) -> None:
    """
    Load the decryption key from the keyserver, decrypt the development key, start sops to edit the plaintext secrets of the provided file
    """

    host_configuration_dir = FLAKE / "nixosConfigurations" / hostname
    encrypted_file = host_configuration_dir / file

    assert host_configuration_dir.is_dir(), f"""
        There is no configuration folder found for host {hostname}.
        Create a nixos configuration at path `{host_configuration_dir.as_posix()}` first!
    """

    if not (
        encrypted_file.name.endswith("encrypted.yaml")
        or encrypted_file.name.endswith("encrypted.json")
    ):
        raise ValueError(
            f"Filename must end with *.encrypted.yaml or *.encrypted.json: {encrypted_file.name}"
        )

    environment = os.environ.copy()
    environment.pop("SOPS_AGE_KEY_FILE", None)
    environment["SOPS_AGE_KEY"] = dev_key_decrypt()

    result = subprocess.run(
        [
            "sops",
            *(["--input-type", "binary", "--output-type", "binary"] if binary else []),
            encrypted_file.as_posix(),
        ],
        cwd=FLAKE,
        env=environment,
        # check=True, # Only verifies exit code 0
    )

    acceptable_sops_exitcodes = [
        0,
        200,  # FileHasNotBeenModified
    ]
    if result.returncode not in acceptable_sops_exitcodes:
        raise subprocess.CalledProcessError(result.returncode, result.args)


@task
# USAGE; invoke ssh-key-create development [-f "secrets.encrypted.yaml"] [-k "ssh_host_ed25519_key"]
def ssh_key_create(
    c: Any,
    hostname: str,
    file: str = "secrets.encrypted.yaml",
    key: str = "ssh_host_ed25519_key",
) -> None:
    """
    Create and encrypt a new SSH private host key.
    """

    host_configuration_dir = FLAKE / "nixosConfigurations" / hostname
    encrypted_file = host_configuration_dir / file

    assert host_configuration_dir.is_dir(), f"""
        There is no configuration folder found for host {hostname}.
        Create a nixos configuration at path `{host_configuration_dir.as_posix()}` first!
    """

    if not encrypted_file.name.endswith("encrypted.yaml"):
        raise ValueError(
            f"Filename must end with *.encrypted.yaml: {encrypted_file.name}"
        )

    if not key:
        raise ValueError("Secret key name cannot be empty")

    with TemporaryDirectory() as tmpdir:
        # Prepare filepath and secure file access to store sensitive key material
        tmp = Path(tmpdir)
        tmp.mkdir(parents=True, exist_ok=True)
        tmp.chmod(0o755)
        host_key = tmp / "ssh_host_ed25519_key"
        pub_host_key = host_key.with_suffix(".pub")

        # Create a new key of type 'ed25519', written to the designated filepath
        #
        # ERROR; Explicit program and argument syntax (list/bracket form), because we're not using
        # the shell as intermediate command interpreter
        subprocess.run(
            [
                "ssh-keygen",
                "-t",
                "ed25519",
                "-N",
                "",  # No password
                "-f",
                host_key.as_posix(),
            ],
            check=True,
        )

        with open(host_key, "r", opener=private_opener) as file_handle:
            ssh_private_key = file_handle.read()

        with open(pub_host_key, "r") as file_handle:
            public_key = file_handle.read()

    if not (ssh_private_key and public_key):
        raise RuntimeError("Generated SSH key files are empty")

    # ERROR; File must exist for 'sops set' to work
    if not encrypted_file.is_file():
        subprocess.run(
            [
                "sops",
                "encrypt",
                "--input-type",
                "json",
                "--output-type",
                "yaml",
                "--output",
                encrypted_file.as_posix(),
                # Input file
                "/dev/stdin",
            ],
            cwd=FLAKE,
            input=json.dumps({key: ssh_private_key}),
            text=True,
            check=True,
        )
    else:
        if find_string_in_file(encrypted_file, f"{key}:"):
            warnings.warn(
                "The secret name is found in the encrypted file, it's very likely we're gonna overwrite existing data"
            )
            if not ask_user_input(
                "Do you want to keep going and possibly overwrite your encrypted data?"
            ):
                raise ValueError("Process canceled as to not overwrite data")

        environment = os.environ.copy()
        environment.pop("SOPS_AGE_KEY_FILE", None)
        environment["SOPS_AGE_KEY"] = dev_key_decrypt()

        subprocess.run(
            [
                "sops",
                "set",
                encrypted_file.as_posix(),
                json.dumps([key]),  # Key name selector
                json.dumps(ssh_private_key),  # Value as json string
            ],
            cwd=FLAKE,
            env=environment,
            check=True,
        )

    print(
        f"Private key succesfully encrypted! Below is the corresponding public key\n{public_key}"
    )


@task
def development_key_create(c: Any, name: str = "development") -> None:
    """
    Creates a new development key, password protect it, and store it at path {FLAKE}/<name>.age

    The public part should be provided to SOPS (see '.sops.yaml') for encryption.
    The private part should be made available, in decrypted form, when deploying secrets.
    """
    c.run(f'rage -p -o "{name}.age" <(rage-keygen)')


@task
# USAGE; invoke decrypter-key-create development [-k "development_decrypter"]
def decrypter_key_create(c: Any, hostname: str, key: str = None) -> None:
    """
    Create a new AGE key for encrypting/decrypting all secrets provided to a host.
    """
    host_configuration_dir = FLAKE / "nixosConfigurations" / hostname
    encrypted_file = host_configuration_dir / decryptor_encrypted_filename_default()

    if not host_configuration_dir.is_dir():
        warnings.warn(
            "There is no configuration folder found for the provided hostname"
        )
        if not ask_user_input(
            f"Do you want to create a folder at path {host_configuration_dir.as_posix()}"
        ):
            raise ValueError(f"No configuration folder for host {hostname}")
        host_configuration_dir.mkdir(exist_ok=True)

    age_key = subprocess.run(
        "rage-keygen",
        text=True,  # stdin/stdout are opened in text mode
        check=True,  # Throw exception if command fails
        capture_output=True,  # Redirect stdout/stderr
    ).stdout.strip()

    if not age_key:
        raise RuntimeError("rage-keygen produced empty output")

    # Print everything except last line (presumably private key) to the terminal
    # for the user to further process.
    print("\n".join(age_key.splitlines()[:-1]))

    if not key:
        key = decryptor_name_default(hostname)
        warnings.warn(f"Defaulting to secret key-name {key}")

    # ERROR; File must exist for 'sops set' to work
    if not encrypted_file.is_file():
        subprocess.run(
            [
                "sops",
                "encrypt",
                "--input-type",
                "json",
                "--output-type",
                "yaml",
                # "--filename-override", # Yes, sops has a very weird CLI -_-
                "--output",
                encrypted_file.as_posix(),
                # Input file
                "/dev/stdin",
            ],
            cwd=FLAKE,
            input=json.dumps({key: age_key}),
            text=True,
            check=True,
        )
        return

    if find_string_in_file(encrypted_file, f"{key}:"):
        warnings.warn(
            "The secret name is found in the encrypted file, it's very likely we're gonna overwrite existing data"
        )
        if not ask_user_input(
            "Do you want to keep going and possibly overwrite your encrypted data?"
        ):
            raise ValueError("Process canceled as to not overwrite data")

    environment = os.environ.copy()
    environment.pop("SOPS_AGE_KEY_FILE", None)
    environment["SOPS_AGE_KEY"] = dev_key_decrypt()

    subprocess.run(
        [
            "sops",
            "set",
            encrypted_file.as_posix(),
            json.dumps([key]),  # Key name selector
            json.dumps(age_key),  # Value as json string
        ],
        cwd=FLAKE,
        env=environment,
        check=True,
    )


@task
def format(c: Any) -> None:
    """
    Use the defined formatter to format all source files in this repository
    """
    subprocess.run(["nix", "fmt", PROJECT_DIR], cwd=FLAKE, check=False)


@task
def documentation(c: Any) -> None:
    """
    Build host and network documentation
    """
    subprocess.run(
        [
            "nix",
            "build",
            f"{DOCS}#topology.x86_64-linux.config.output",
            "--override-input",
            "configuration",
            FLAKE,
        ],
        check=True,
    )