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

enable ASYNC, DTZ, YTT and EM lints

Browse Source
This commit is contained in:
Jörg Thalheim
2024-09-02 13:55:46 +02:00
parent d5440594be
commit 15ff74f7c2
98 changed files with 526 additions and 421 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
pkgs/clan-cli/clan_cli/api/__init__.py
View File

@@ -85,8 +85,7 @@ class MethodRegistry:
def register_abstract(self, fn: Callable[..., T]) -> Callable[..., T]:
@wraps(fn)
def wrapper(*args: Any, op_key: str, **kwargs: Any) -> ApiResponse[T]:
raise NotImplementedError(
f"""{fn.__name__} - The platform didn't implement this function.
msg = f"""{fn.__name__} - The platform didn't implement this function.
---
# Example
@@ -103,16 +102,18 @@ def open_file(file_request: FileRequest) -> str | None:
API.register(open_file)
---
"""
)
raise NotImplementedError(msg)
self.register(wrapper)
return fn
def register(self, fn: Callable[..., T]) -> Callable[..., T]:
if fn.__name__ in self._registry:
raise ValueError(f"Function {fn.__name__} already registered")
msg = f"Function {fn.__name__} already registered"
raise ValueError(msg)
if fn.__name__ in self._orig_signature:
raise ValueError(f"Function {fn.__name__} already registered")
msg = f"Function {fn.__name__} already registered"
raise ValueError(msg)
# make copy of original function
self._orig_signature[fn.__name__] = signature(fn)

6
pkgs/clan-cli/clan_cli/api/admin.py
View File

@@ -38,12 +38,14 @@ def set_admin_service(
inventory = load_inventory_eval(base_url)
if not allowed_keys:
raise ValueError("At least one key must be provided to ensure access")
msg = "At least one key must be provided to ensure access"
raise ValueError(msg)
keys = {}
for name, keyfile in allowed_keys.items():
if not keyfile.startswith("/"):
raise ValueError(f"Keyfile '{keyfile}' must be an absolute path")
msg = f"Keyfile '{keyfile}' must be an absolute path"
raise ValueError(msg)
with open(keyfile) as f:
pubkey = f.read()
keys[name] = pubkey

20
pkgs/clan-cli/clan_cli/api/modules.py
View File

@@ -51,8 +51,9 @@ def extract_frontmatter(readme_content: str, err_scope: str) -> tuple[Frontmatte
# Parse the TOML frontmatter
frontmatter_parsed = tomllib.loads(frontmatter_raw)
except tomllib.TOMLDecodeError as e:
msg = f"Error parsing TOML frontmatter: {e}"
raise ClanError(
f"Error parsing TOML frontmatter: {e}",
msg,
description=f"Invalid TOML frontmatter. {err_scope}",
location="extract_frontmatter",
) from e
@@ -60,8 +61,9 @@ def extract_frontmatter(readme_content: str, err_scope: str) -> tuple[Frontmatte
return Frontmatter(**frontmatter_parsed), remaining_content
# If no frontmatter is found, raise an error
msg = "Invalid README: Frontmatter not found."
raise ClanError(
"Invalid README: Frontmatter not found.",
msg,
location="extract_frontmatter",
description=f"{err_scope} does not contain valid frontmatter.",
)
@@ -98,8 +100,9 @@ def get_modules(base_path: str) -> dict[str, str]:
proc = run_no_stdout(cmd)
res = proc.stdout.strip()
except ClanCmdError as e:
msg = "clanInternals might not have clanModules attributes"
raise ClanError(
"clanInternals might not have clanModules attributes",
msg,
location=f"list_modules {base_path}",
description="Evaluation failed on clanInternals.clanModules attribute",
) from e
@@ -127,15 +130,17 @@ def get_module_info(
Retrieves information about a module
"""
if not module_path:
msg = "Module not found"
raise ClanError(
"Module not found",
msg,
location=f"show_module_info {module_name}",
description="Module does not exist",
)
module_readme = Path(module_path) / "README.md"
if not module_readme.exists():
msg = "Module not found"
raise ClanError(
"Module not found",
msg,
location=f"show_module_info {module_name}",
description="Module does not exist or doesn't have any README.md file",
)
@@ -170,9 +175,8 @@ def set_service_instance(
service_keys = get_type_hints(Service).keys()
if module_name not in service_keys:
raise ValueError(
f"{module_name} is not a valid Service attribute. Expected one of {', '.join(service_keys)}."
)
msg = f"{module_name} is not a valid Service attribute. Expected one of {', '.join(service_keys)}."
raise ValueError(msg)
inventory = load_inventory_json(base_path)
target_type = get_args(get_type_hints(Service)[module_name])[1]

33
pkgs/clan-cli/clan_cli/api/serde.py
View File

@@ -129,7 +129,8 @@ def construct_value(
if loc is None:
loc = []
if t is None and field_value:
raise ClanError(f"Expected None but got: {field_value}", location=f"{loc}")
msg = f"Expected None but got: {field_value}"
raise ClanError(msg, location=f"{loc}")
if is_type_in_union(t, type(None)) and field_value is None:
# Sometimes the field value is None, which is valid if the type hint allows None
@@ -145,8 +146,11 @@ def construct_value(
# Field_value must be a string
elif is_type_in_union(t, Path):
if not isinstance(field_value, str):
msg = (
f"Expected string, cannot construct pathlib.Path() from: {field_value} "
)
raise ClanError(
f"Expected string, cannot construct pathlib.Path() from: {field_value} ",
msg,
location=f"{loc}",
)
@@ -155,7 +159,8 @@ def construct_value(
# Trivial values
elif t is str:
if not isinstance(field_value, str):
raise ClanError(f"Expected string, got {field_value}", location=f"{loc}")
msg = f"Expected string, got {field_value}"
raise ClanError(msg, location=f"{loc}")
return field_value
@@ -178,7 +183,8 @@ def construct_value(
# dict
elif get_origin(t) is list:
if not isinstance(field_value, list):
raise ClanError(f"Expected list, got {field_value}", location=f"{loc}")
msg = f"Expected list, got {field_value}"
raise ClanError(msg, location=f"{loc}")
return [construct_value(get_args(t)[0], item) for item in field_value]
elif get_origin(t) is dict and isinstance(field_value, dict):
@@ -189,9 +195,8 @@ def construct_value(
elif get_origin(t) is Literal:
valid_values = get_args(t)
if field_value not in valid_values:
raise ClanError(
f"Expected one of {valid_values}, got {field_value}", location=f"{loc}"
)
msg = f"Expected one of {valid_values}, got {field_value}"
raise ClanError(msg, location=f"{loc}")
return field_value
elif get_origin(t) is Annotated:
@@ -202,7 +207,8 @@ def construct_value(
# Unhandled
else:
raise ClanError(f"Unhandled field type {t} with value {field_value}")
msg = f"Unhandled field type {t} with value {field_value}"
raise ClanError(msg)
def construct_dataclass(
@@ -215,7 +221,8 @@ def construct_dataclass(
if path is None:
path = []
if not is_dataclass(t):
raise ClanError(f"{t.__name__} is not a dataclass")
msg = f"{t.__name__} is not a dataclass"
raise ClanError(msg)
# Attempt to create an instance of the data_class#
field_values: dict[str, Any] = {}
@@ -251,9 +258,8 @@ def construct_dataclass(
for field_name in required:
if field_name not in field_values:
formatted_path = " ".join(path)
raise ClanError(
f"Default value missing for: '{field_name}' in {t} {formatted_path}, got Value: {data}"
)
msg = f"Default value missing for: '{field_name}' in {t} {formatted_path}, got Value: {data}"
raise ClanError(msg)
return t(**field_values) # type: ignore
@@ -265,7 +271,8 @@ def from_dict(
path = []
if is_dataclass(t):
if not isinstance(data, dict):
raise ClanError(f"{data} is not a dict. Expected {t}")
msg = f"{data} is not a dict. Expected {t}"
raise ClanError(msg)
return construct_dataclass(t, data, path) # type: ignore
else:
return construct_value(t, data, path)

27
pkgs/clan-cli/clan_cli/api/util.py
View File

@@ -122,9 +122,8 @@ def type_to_dict(
# And return the resolved type instead of the TypeVar
resolved = type_map.get(t)
if not resolved:
raise JSchemaTypeError(
f"{scope} - TypeVar {t} not found in type_map, map: {type_map}"
)
msg = f"{scope} - TypeVar {t} not found in type_map, map: {type_map}"
raise JSchemaTypeError(msg)
return type_to_dict(type_map.get(t), scope, type_map)
elif hasattr(t, "__origin__"): # Check if it's a generic type
@@ -134,7 +133,8 @@ def type_to_dict(
if origin is None:
# Non-generic user-defined or built-in type
# TODO: handle custom types
raise JSchemaTypeError(f"{scope} Unhandled Type: ", origin)
msg = f"{scope} Unhandled Type: "
raise JSchemaTypeError(msg, origin)
elif origin is Literal:
# Handle Literal values for enums in JSON Schema
@@ -179,7 +179,8 @@ def type_to_dict(
new_map.update(inspect_dataclass_fields(t))
return type_to_dict(origin, scope, new_map)
raise JSchemaTypeError(f"{scope} - Error api type not yet supported {t!s}")
msg = f"{scope} - Error api type not yet supported {t!s}"
raise JSchemaTypeError(msg)
elif isinstance(t, type):
if t is str:
@@ -193,23 +194,23 @@ def type_to_dict(
if t is object:
return {"type": "object"}
if t is Any:
raise JSchemaTypeError(
f"{scope} - Usage of the Any type is not supported for API functions. In: {scope}"
)
msg = f"{scope} - Usage of the Any type is not supported for API functions. In: {scope}"
raise JSchemaTypeError(msg)
if t is pathlib.Path:
return {
# TODO: maybe give it a pattern for URI
"type": "string",
}
if t is dict:
raise JSchemaTypeError(
f"{scope} - Generic 'dict' type not supported. Use dict[str, Any] or any more expressive type."
)
msg = f"{scope} - Generic 'dict' type not supported. Use dict[str, Any] or any more expressive type."
raise JSchemaTypeError(msg)
# Optional[T] gets internally transformed Union[T,NoneType]
if t is NoneType:
return {"type": "null"}
raise JSchemaTypeError(f"{scope} - Error primitive type not supported {t!s}")
msg = f"{scope} - Error primitive type not supported {t!s}"
raise JSchemaTypeError(msg)
else:
raise JSchemaTypeError(f"{scope} - Error type not supported {t!s}")
msg = f"{scope} - Error type not supported {t!s}"
raise JSchemaTypeError(msg)

12
pkgs/clan-cli/clan_cli/backups/create.py
View File

@@ -22,24 +22,28 @@ def create_backup(machine: Machine, provider: str | None = None) -> None:
[backup_scripts["providers"][provider]["create"]],
)
if proc.returncode != 0:
raise ClanError("failed to start backup")
msg = "failed to start backup"
raise ClanError(msg)
else:
print("successfully started backup")
else:
if provider not in backup_scripts["providers"]:
raise ClanError(f"provider {provider} not found")
msg = f"provider {provider} not found"
raise ClanError(msg)
proc = machine.target_host.run(
[backup_scripts["providers"][provider]["create"]],
)
if proc.returncode != 0:
raise ClanError("failed to start backup")
msg = "failed to start backup"
raise ClanError(msg)
else:
print("successfully started backup")
def create_command(args: argparse.Namespace) -> None:
if args.flake is None:
raise ClanError("Could not find clan flake toplevel directory")
msg = "Could not find clan flake toplevel directory"
raise ClanError(msg)
machine = Machine(name=args.machine, flake=args.flake)
create_backup(machine=machine, provider=args.provider)

3
pkgs/clan-cli/clan_cli/backups/list.py
View File

@@ -54,7 +54,8 @@ def list_backups(machine: Machine, provider: str | None = None) -> list[Backup]:
def list_command(args: argparse.Namespace) -> None:
if args.flake is None:
raise ClanError("Could not find clan flake toplevel directory")
msg = "Could not find clan flake toplevel directory"
raise ClanError(msg)
machine = Machine(name=args.machine, flake=args.flake)
backups = list_backups(machine=machine, provider=args.provider)
for backup in backups:

18
pkgs/clan-cli/clan_cli/backups/restore.py
View File

@@ -32,9 +32,8 @@ def restore_service(machine: Machine, name: str, provider: str, service: str) ->
extra_env=env,
)
if proc.returncode != 0:
raise ClanError(
f"failed to run preRestoreCommand: {pre_restore}, error was: {proc.stdout}"
)
msg = f"failed to run preRestoreCommand: {pre_restore}, error was: {proc.stdout}"
raise ClanError(msg)
proc = machine.target_host.run(
[backup_metadata["providers"][provider]["restore"]],
@@ -42,9 +41,8 @@ def restore_service(machine: Machine, name: str, provider: str, service: str) ->
extra_env=env,
)
if proc.returncode != 0:
raise ClanError(
f"failed to restore backup: {backup_metadata['providers'][provider]['restore']}"
)
msg = f"failed to restore backup: {backup_metadata['providers'][provider]['restore']}"
raise ClanError(msg)
if post_restore := backup_folders[service]["postRestoreCommand"]:
proc = machine.target_host.run(
@@ -53,9 +51,8 @@ def restore_service(machine: Machine, name: str, provider: str, service: str) ->
extra_env=env,
)
if proc.returncode != 0:
raise ClanError(
f"failed to run postRestoreCommand: {post_restore}, error was: {proc.stdout}"
)
msg = f"failed to run postRestoreCommand: {post_restore}, error was: {proc.stdout}"
raise ClanError(msg)
def restore_backup(
@@ -85,7 +82,8 @@ def restore_backup(
def restore_command(args: argparse.Namespace) -> None:
if args.flake is None:
raise ClanError("Could not find clan flake toplevel directory")
msg = "Could not find clan flake toplevel directory"
raise ClanError(msg)
machine = Machine(name=args.machine, flake=args.flake)
restore_backup(
machine=machine,

5
pkgs/clan-cli/clan_cli/clan/inspect.py
View File

@@ -50,9 +50,8 @@ def inspect_flake(flake_url: str | Path, machine_name: str) -> FlakeConfig:
# Check if the machine exists
machines: list[str] = list_nixos_machines(flake_url)
if machine_name not in machines:
raise ClanError(
f"Machine {machine_name} not found in {flake_url}. Available machines: {', '.join(machines)}"
)
msg = f"Machine {machine_name} not found in {flake_url}. Available machines: {', '.join(machines)}"
raise ClanError(msg)
machine = Machine(machine_name, FlakeId(str(flake_url)))
vm = inspect_vm(machine)

9
pkgs/clan-cli/clan_cli/clan/show.py
View File

@@ -27,8 +27,9 @@ def show_clan_meta(uri: str | Path) -> Meta:
proc = run_no_stdout(cmd)
res = proc.stdout.strip()
except ClanCmdError as e:
msg = "Evaluation failed on meta attribute"
raise ClanError(
"Evaluation failed on meta attribute",
msg,
location=f"show_clan {uri}",
description=str(e.cmd),
) from e
@@ -45,8 +46,9 @@ def show_clan_meta(uri: str | Path) -> Meta:
icon_path = meta_icon
elif scheme in [""]:
if Path(meta_icon).is_absolute():
msg = "Invalid absolute path"
raise ClanError(
"Invalid absolute path",
msg,
location=f"show_clan {uri}",
description="Icon path must be a URL or a relative path.",
)
@@ -54,8 +56,9 @@ def show_clan_meta(uri: str | Path) -> Meta:
else:
icon_path = str((Path(uri) / meta_icon).resolve())
else:
msg = "Invalid schema"
raise ClanError(
"Invalid schema",
msg,
location=f"show_clan {uri}",
description="Icon path must be a URL or a relative path.",
)

3
pkgs/clan-cli/clan_cli/clan_uri.py
View File

@@ -74,7 +74,8 @@ class ClanURI:
if uri.startswith("clan://"):
nested_uri = uri[7:]
else:
raise ClanError(f"Invalid uri: expected clan://, got {uri}")
msg = f"Invalid uri: expected clan://, got {uri}"
raise ClanError(msg)
# Parse the URI into components
# url://netloc/path;parameters?query#fragment

7
pkgs/clan-cli/clan_cli/cmd.py
View File

@@ -1,3 +1,4 @@
import datetime
import logging
import os
import select
@@ -5,7 +6,7 @@ import shlex
import subprocess
import sys
import weakref
from datetime import datetime, timedelta
from datetime import timedelta
from enum import Enum
from pathlib import Path
from typing import IO, Any
@@ -116,7 +117,7 @@ def run(
)
else:
glog.debug(f"$: {shlex.join(cmd)} \nCaller: {get_caller()}")
tstart = datetime.now()
tstart = datetime.datetime.now(tz=datetime.UTC)
# Start the subprocess
process = subprocess.Popen(
@@ -132,7 +133,7 @@ def run(
process.communicate(input)
else:
process.wait()
tend = datetime.now()
tend = datetime.datetime.now(tz=datetime.UTC)
global TIME_TABLE
TIME_TABLE.add(shlex.join(cmd), tend - tstart)

22
pkgs/clan-cli/clan_cli/config/__init__.py
View File

@@ -40,7 +40,8 @@ def map_type(nix_type: str) -> Any:
subtype = nix_type.removeprefix("list of ")
return list[map_type(subtype)] # type: ignore
else:
raise ClanError(f"Unknown type {nix_type}")
msg = f"Unknown type {nix_type}"
raise ClanError(msg)
# merge two dicts recursively
@@ -79,7 +80,8 @@ def cast(value: Any, input_type: Any, opt_description: str) -> Any:
elif value[0] in ["false", "False", "no", "n", "0"]:
return False
else:
raise ClanError(f"Invalid value {value} for boolean")
msg = f"Invalid value {value} for boolean"
raise ClanError(msg)
# handle lists
elif get_origin(input_type) is list:
subtype = input_type.__args__[0]
@@ -87,9 +89,8 @@ def cast(value: Any, input_type: Any, opt_description: str) -> Any:
# handle dicts
elif get_origin(input_type) is dict:
if not isinstance(value, dict):
raise ClanError(
f"Cannot set {opt_description} directly. Specify a suboption like {opt_description}.<name>"
)
msg = f"Cannot set {opt_description} directly. Specify a suboption like {opt_description}.<name>"
raise ClanError(msg)
subtype = input_type.__args__[1]
return {k: cast(v, subtype, opt_description) for k, v in value.items()}
elif str(input_type) == "str | None":
@@ -98,12 +99,12 @@ def cast(value: Any, input_type: Any, opt_description: str) -> Any:
return value[0]
else:
if len(value) > 1:
raise ClanError(f"Too many values for {opt_description}")
msg = f"Too many values for {opt_description}"
raise ClanError(msg)
return input_type(value[0])
except ValueError as e:
raise ClanError(
f"Invalid type for option {opt_description} (expected {input_type.__name__})"
) from e
msg = f"Invalid type for option {opt_description} (expected {input_type.__name__})"
raise ClanError(msg) from e
def options_for_machine(
@@ -237,7 +238,8 @@ def find_option(
# element of the option path and find matching parent option
# (see examples above for why this is needed)
if len(option_path) == 1:
raise ClanError(f"Option {option_description} not found")
msg = f"Option {option_description} not found"
raise ClanError(msg)
option_path_parent = option_path[:-1]
attr_prefix = option_path[-1]
return find_option(

3
pkgs/clan-cli/clan_cli/config/machine.py
View File

@@ -89,7 +89,8 @@ def config_for_machine(flake_dir: Path, machine_name: str) -> dict:
def set_config_for_machine(flake_dir: Path, machine_name: str, config: dict) -> None:
hostname_regex = r"^(?!-)[A-Za-z0-9-]{1,63}(?<!-)$"
if not re.match(hostname_regex, machine_name):
raise ClanError("Machine name must be a valid hostname")
msg = "Machine name must be a valid hostname"
raise ClanError(msg)
if "networking" in config and "hostName" in config["networking"]:
if machine_name != config["networking"]["hostName"]:
raise ClanHttpError(

26
pkgs/clan-cli/clan_cli/config/parsing.py
View File

@@ -42,12 +42,15 @@ def subtype_from_schema(schema: dict[str, Any]) -> type:
sub_type = subtype_from_schema(schema["additionalProperties"])
return dict[str, sub_type] # type: ignore
elif "properties" in schema:
raise ClanError("Nested dicts are not supported")
msg = "Nested dicts are not supported"
raise ClanError(msg)
else:
raise ClanError("Unknown object type")
msg = "Unknown object type"
raise ClanError(msg)
elif schema["type"] == "array":
if "items" not in schema:
raise ClanError("Untyped arrays are not supported")
msg = "Untyped arrays are not supported"
raise ClanError(msg)
sub_type = subtype_from_schema(schema["items"])
return list[sub_type] # type: ignore
else:
@@ -71,9 +74,11 @@ def type_from_schema_path(
subtype = type_from_schema_path(schema["additionalProperties"], path[1:])
return subtype
else:
raise ClanError(f"Unknown type for path {path}")
msg = f"Unknown type for path {path}"
raise ClanError(msg)
else:
raise ClanError(f"Unknown type for path {path}")
msg = f"Unknown type for path {path}"
raise ClanError(msg)
def options_types_from_schema(schema: dict[str, Any]) -> dict[str, type]:
@@ -86,9 +91,8 @@ def options_types_from_schema(schema: dict[str, Any]) -> dict[str, type]:
if "additionalProperties" in value:
sub_type = value["additionalProperties"].get("type")
if sub_type not in type_map:
raise ClanError(
f"Unsupported object type {sub_type} (field {name})"
)
msg = f"Unsupported object type {sub_type} (field {name})"
raise ClanError(msg)
result[f"{name}.<name>"] = type_map[sub_type]
continue
# handle properties
@@ -98,10 +102,12 @@ def options_types_from_schema(schema: dict[str, Any]) -> dict[str, type]:
continue
elif type_ == "array":
if "items" not in value:
raise ClanError(f"Untyped arrays are not supported (field: {name})")
msg = f"Untyped arrays are not supported (field: {name})"
raise ClanError(msg)
sub_type = value["items"].get("type")
if sub_type not in type_map:
raise ClanError(f"Unsupported list type {sub_type} (field {name})")
msg = f"Unsupported list type {sub_type} (field {name})"
raise ClanError(msg)
sub_type_: type = type_map[sub_type]
result[name] = list[sub_type_] # type: ignore
continue

3
pkgs/clan-cli/clan_cli/config/schema.py
View File

@@ -110,5 +110,6 @@ def machine_schema(
env=env,
)
if proc.returncode != 0:
raise ClanError(f"Failed to read schema:\n{proc.stderr}")
msg = f"Failed to read schema:\n{proc.stderr}"
raise ClanError(msg)
return json.loads(proc.stdout)

5
pkgs/clan-cli/clan_cli/facts/generate.py
View File

@@ -161,9 +161,8 @@ def _generate_facts_for_machine(
if service and service not in machine.facts_data:
services = list(machine.facts_data.keys())
raise ClanError(
f"Could not find service with name: {service}. The following services are available: {services}"
)
msg = f"Could not find service with name: {service}. The following services are available: {services}"
raise ClanError(msg)
if service:
machine_service_facts = {service: machine.facts_data[service]}

5
pkgs/clan-cli/clan_cli/facts/public_modules/in_repo.py
View File

@@ -25,9 +25,8 @@ class FactStore(FactStoreBase):
fact_path.write_bytes(value)
return fact_path
else:
raise ClanError(
f"in_flake fact storage is only supported for local flakes: {self.machine.flake}"
)
msg = f"in_flake fact storage is only supported for local flakes: {self.machine.flake}"
raise ClanError(msg)
def exists(self, service: str, name: str) -> bool:
fact_path = (

3
pkgs/clan-cli/clan_cli/facts/public_modules/vm.py
View File

@@ -32,7 +32,8 @@ class FactStore(FactStoreBase):
fact_path = self.dir / service / name
if fact_path.exists():
return fact_path.read_bytes()
raise ClanError(f"Fact {name} for service {service} not found")
msg = f"Fact {name} for service {service} not found"
raise ClanError(msg)
# get all facts
def get_all(self) -> dict[str, dict[str, bytes]]:

22
pkgs/clan-cli/clan_cli/flash.py
View File

@@ -44,7 +44,8 @@ def list_possible_keymaps() -> list[str]:
keymaps_dir = Path(result.stdout.strip()) / "share" / "keymaps"
if not keymaps_dir.exists():
raise FileNotFoundError(f"Keymaps directory '{keymaps_dir}' does not exist.")
msg = f"Keymaps directory '{keymaps_dir}' does not exist."
raise FileNotFoundError(msg)
keymap_files = []
@@ -65,7 +66,8 @@ def list_possible_languages() -> list[str]:
locale_file = Path(result.stdout.strip()) / "share" / "i18n" / "SUPPORTED"
if not locale_file.exists():
raise FileNotFoundError(f"Locale file '{locale_file}' does not exist.")
msg = f"Locale file '{locale_file}' does not exist."
raise FileNotFoundError(msg)
with locale_file.open() as f:
lines = f.readlines()
@@ -107,18 +109,20 @@ def flash_machine(
if system_config.language:
if system_config.language not in list_possible_languages():
raise ClanError(
msg = (
f"Language '{system_config.language}' is not a valid language. "
f"Run 'clan flash --list-languages' to see a list of possible languages."
)
raise ClanError(msg)
system_config_nix["i18n"] = {"defaultLocale": system_config.language}
if system_config.keymap:
if system_config.keymap not in list_possible_keymaps():
raise ClanError(
msg = (
f"Keymap '{system_config.keymap}' is not a valid keymap. "
f"Run 'clan flash --list-keymaps' to see a list of possible keymaps."
)
raise ClanError(msg)
system_config_nix["console"] = {"keyMap": system_config.keymap}
if system_config.ssh_keys_path:
@@ -127,9 +131,8 @@ def flash_machine(
try:
root_keys.append(key_path.read_text())
except OSError as e:
raise ClanError(
f"Cannot read SSH public key file: {key_path}: {e}"
) from e
msg = f"Cannot read SSH public key file: {key_path}: {e}"
raise ClanError(msg) from e
system_config_nix["users"] = {
"users": {"root": {"openssh": {"authorizedKeys": {"keys": root_keys}}}}
}
@@ -153,9 +156,8 @@ def flash_machine(
if os.geteuid() != 0:
if shutil.which("sudo") is None:
raise ClanError(
"sudo is required to run disko-install as a non-root user"
)
msg = "sudo is required to run disko-install as a non-root user"
raise ClanError(msg)
wrapper = 'set -x; disko_install=$(command -v disko-install); exec sudo "$disko_install" "$@"'
disko_install.extend(["bash", "-c", wrapper])

3
pkgs/clan-cli/clan_cli/git.py
View File

@@ -32,7 +32,8 @@ def commit_files(
# check that the file is in the git repository
for file_path in file_paths:
if not Path(file_path).resolve().is_relative_to(repo_dir.resolve()):
raise ClanError(f"File {file_path} is not in the git repository {repo_dir}")
msg = f"File {file_path} is not in the git repository {repo_dir}"
raise ClanError(msg)
# generate commit message if not provided
if commit_message is None:
commit_message = ""

7
pkgs/clan-cli/clan_cli/history/add.py
View File

@@ -54,7 +54,8 @@ def list_history() -> list[HistoryEntry]:
parsed[i] = _merge_dicts(p, p.get("settings", {}))
logs = [HistoryEntry(**p) for p in parsed]
except (json.JSONDecodeError, TypeError) as ex:
raise ClanError(f"History file at {user_history_file()} is corrupted") from ex
msg = f"History file at {user_history_file()} is corrupted"
raise ClanError(msg) from ex
return logs
@@ -63,7 +64,7 @@ def new_history_entry(url: str, machine: str) -> HistoryEntry:
flake = inspect_flake(url, machine)
return HistoryEntry(
flake=flake,
last_used=datetime.datetime.now().isoformat(),
last_used=datetime.datetime.now(tz=datetime.UTC).isoformat(),
)
@@ -93,7 +94,7 @@ def _add_maschine_to_history_list(
new_entry.flake.flake_url == str(uri_path)
and new_entry.flake.flake_attr == uri_machine
):
new_entry.last_used = datetime.datetime.now().isoformat()
new_entry.last_used = datetime.datetime.now(tz=datetime.UTC).isoformat()
return new_entry
new_entry = new_history_entry(uri_path, uri_machine)

3
pkgs/clan-cli/clan_cli/history/update.py
View File

@@ -33,7 +33,8 @@ def update_history() -> list[HistoryEntry]:
flake = inspect_flake(uri.get_url(), uri.machine_name)
flake.flake_url = flake.flake_url
entry = HistoryEntry(
flake=flake, last_used=datetime.datetime.now().isoformat()
flake=flake,
last_used=datetime.datetime.now(tz=datetime.UTC).isoformat(),
)
write_history_file(logs)

6
pkgs/clan-cli/clan_cli/inventory/__init__.py
View File

@@ -114,7 +114,8 @@ def load_inventory_eval(flake_dir: str | Path) -> Inventory:
inventory = from_dict(Inventory, data)
return inventory
except json.JSONDecodeError as e:
raise ClanError(f"Error decoding inventory from flake: {e}") from e
msg = f"Error decoding inventory from flake: {e}"
raise ClanError(msg) from e
def load_inventory_json(
@@ -134,7 +135,8 @@ def load_inventory_json(
inventory = from_dict(Inventory, res)
except json.JSONDecodeError as e:
# Error decoding the inventory file
raise ClanError(f"Error decoding inventory file: {e}") from e
msg = f"Error decoding inventory file: {e}"
raise ClanError(msg) from e
if not inventory_file.exists():
# Copy over the meta from the flake if the inventory is not initialized

8
pkgs/clan-cli/clan_cli/machines/create.py
View File

@@ -20,16 +20,16 @@ log = logging.getLogger(__name__)
def create_machine(flake: FlakeId, machine: Machine) -> None:
hostname_regex = r"^(?!-)[A-Za-z0-9-]{1,63}(?<!-)$"
if not re.match(hostname_regex, machine.name):
raise ClanError(
"Machine name must be a valid hostname", location="Create Machine"
)
msg = "Machine name must be a valid hostname"
raise ClanError(msg, location="Create Machine")
inventory = load_inventory_json(flake.path)
full_inventory = load_inventory_eval(flake.path)
if machine.name in full_inventory.machines.keys():
raise ClanError(f"Machine with the name {machine.name} already exists")
msg = f"Machine with the name {machine.name} already exists"
raise ClanError(msg)
print(f"Define machine {machine.name}", machine)

3
pkgs/clan-cli/clan_cli/machines/delete.py
View File

@@ -15,7 +15,8 @@ def delete_machine(flake: FlakeId, name: str) -> None:
machine = inventory.machines.pop(name, None)
if machine is None:
raise ClanError(f"Machine {name} does not exist")
msg = f"Machine {name} does not exist"
raise ClanError(msg)
save_inventory(inventory, flake.path, f"Delete machine {name}")

9
pkgs/clan-cli/clan_cli/machines/hardware.py
View File

@@ -159,7 +159,8 @@ def generate_machine_hardware_info(
if out.returncode != 0:
log.error(f"Failed to inspect {machine_name}. Address: {hostname}")
log.error(out)
raise ClanError(f"Failed to inspect {machine_name}. Address: {hostname}")
msg = f"Failed to inspect {machine_name}. Address: {hostname}"
raise ClanError(msg)
hw_file = Path(
f"{clan_dir}/machines/{machine_name}/{hw_nix_file if report_type == 'nixos-generate-config' else facter_file}"
@@ -170,8 +171,9 @@ def generate_machine_hardware_info(
is_template = hw_file.exists() and "throw" in hw_file.read_text()
if hw_file.exists() and not force and not is_template:
msg = "File exists."
raise ClanError(
"File exists.",
msg,
description="Hardware file already exists. To force overwrite the existing configuration use '--force'.",
location=f"{__name__} {hw_file}",
)
@@ -205,8 +207,9 @@ def generate_machine_hardware_info(
backup_file.replace(hw_file)
# TODO: Undo the commit
msg = "Invalid hardware-configuration.nix file"
raise ClanError(
"Invalid hardware-configuration.nix file",
msg,
description="The hardware-configuration.nix file is invalid. Please check the file and try again.",
location=f"{__name__} {hw_file}",
) from e

8
pkgs/clan-cli/clan_cli/machines/install.py
View File

@@ -132,7 +132,8 @@ def install_command(args: argparse.Namespace) -> None:
json_ssh_deploy = json.loads(qrcode_scan(args.png))
if not json_ssh_deploy and not args.target_host:
raise ClanError("No target host provided, please provide a target host.")
msg = "No target host provided, please provide a target host."
raise ClanError(msg)
if json_ssh_deploy:
target_host = f"root@{find_reachable_host_from_deploy_json(json_ssh_deploy)}"
@@ -171,13 +172,12 @@ def find_reachable_host_from_deploy_json(deploy_json: dict[str, str]) -> str:
host = addr
break
if not host:
raise ClanError(
f"""
msg = f"""
Could not reach any of the host addresses provided in the json string.
Please doublecheck if they are reachable from your machine.
Try `ping [ADDR]` with one of the addresses: {deploy_json['addrs']}
"""
)
raise ClanError(msg)
return host

12
pkgs/clan-cli/clan_cli/machines/list.py
View File

@@ -44,7 +44,8 @@ def get_inventory_machine_details(
inventory = load_inventory_eval(flake_url)
machine = inventory.machines.get(machine_name)
if machine is None:
raise ClanError(f"Machine {machine_name} not found in inventory")
msg = f"Machine {machine_name} not found in inventory"
raise ClanError(msg)
hw_config_path = (
Path(flake_url) / "machines" / Path(machine_name) / "hardware-configuration.nix"
@@ -73,7 +74,8 @@ def list_nixos_machines(flake_url: str | Path) -> list[str]:
data = json.loads(res)
return data
except json.JSONDecodeError as e:
raise ClanError(f"Error decoding machines from flake: {e}") from e
msg = f"Error decoding machines from flake: {e}"
raise ClanError(msg) from e
@dataclass
@@ -88,12 +90,14 @@ def check_machine_online(
) -> Literal["Online", "Offline"]:
machine = load_inventory_eval(flake_url).machines.get(machine_name)
if not machine:
raise ClanError(f"Machine {machine_name} not found in inventory")
msg = f"Machine {machine_name} not found in inventory"
raise ClanError(msg)
hostname = machine.deploy.targetHost
if not hostname:
raise ClanError(f"Machine {machine_name} does not specify a targetHost")
msg = f"Machine {machine_name} does not specify a targetHost"
raise ClanError(msg)
timeout = opts.timeout if opts and opts.timeout else 20

12
pkgs/clan-cli/clan_cli/machines/machines.py
View File

@@ -102,7 +102,8 @@ class Machine:
elif self.flake.is_remote():
return Path(nix_metadata(self.flake.url)["path"])
else:
raise ClanError(f"Unsupported flake url: {self.flake}")
msg = f"Unsupported flake url: {self.flake}"
raise ClanError(msg)
@property
def target_host(self) -> Host:
@@ -210,7 +211,8 @@ class Machine:
outpath = run_no_stdout(nix_build(args)).stdout.strip()
return Path(outpath)
else:
raise ValueError(f"Unknown method {method}")
msg = f"Unknown method {method}"
raise ValueError(msg)
def eval_nix(
self,
@@ -234,7 +236,8 @@ class Machine:
self._eval_cache[attr] = output
return output
else:
raise ClanError("eval_nix returned not a string")
msg = "eval_nix returned not a string"
raise ClanError(msg)
def build_nix(
self,
@@ -259,4 +262,5 @@ class Machine:
self._build_cache[attr] = output
return output
else:
raise ClanError("build_nix returned not a Path")
msg = "build_nix returned not a Path"
raise ClanError(msg)

10
pkgs/clan-cli/clan_cli/machines/types.py
View File

@@ -12,11 +12,9 @@ def validate_hostname(hostname: str) -> bool:
def machine_name_type(arg_value: str) -> str:
if len(arg_value) > 63:
raise argparse.ArgumentTypeError(
"Machine name must be less than 63 characters long"
)
msg = "Machine name must be less than 63 characters long"
raise argparse.ArgumentTypeError(msg)
if not VALID_HOSTNAME.match(arg_value):
raise argparse.ArgumentTypeError(
"Invalid character in machine name. Allowed characters are a-z, 0-9, ., and -. Must not start with a number"
)
msg = "Invalid character in machine name. Allowed characters are a-z, 0-9, ., and -. Must not start with a number"
raise argparse.ArgumentTypeError(msg)
return arg_value

11
pkgs/clan-cli/clan_cli/machines/update.py
View File

@@ -80,9 +80,8 @@ def upload_sources(
try:
return json.loads(proc.stdout)["path"]
except (json.JSONDecodeError, OSError) as e:
raise ClanError(
f"failed to parse output of {shlex.join(cmd)}: {e}\nGot: {proc.stdout}"
) from e
msg = f"failed to parse output of {shlex.join(cmd)}: {e}\nGot: {proc.stdout}"
raise ClanError(msg) from e
@API.register
@@ -96,7 +95,8 @@ def update_machines(base_path: str, machines: list[InventoryMachine]) -> None:
flake=FlakeId(base_path),
)
if not machine.deploy.targetHost:
raise ClanError(f"'TargetHost' is not set for machine '{machine.name}'")
msg = f"'TargetHost' is not set for machine '{machine.name}'"
raise ClanError(msg)
# Copy targetHost to machine
m.target_host_address = machine.deploy.targetHost
group_machines.append(m)
@@ -161,7 +161,8 @@ def deploy_machine(machines: MachineGroup) -> None:
def update(args: argparse.Namespace) -> None:
if args.flake is None:
raise ClanError("Could not find clan flake toplevel directory")
msg = "Could not find clan flake toplevel directory"
raise ClanError(msg)
machines = []
if len(args.machines) == 1 and args.target_host is not None:
machine = Machine(

3
pkgs/clan-cli/clan_cli/nix/__init__.py
View File

@@ -145,7 +145,8 @@ class Programs:
def run_cmd(programs: list[str], cmd: list[str]) -> list[str]:
for program in programs:
if not Programs.is_allowed(program):
raise ValueError(f"Program not allowed: {program}")
msg = f"Program not allowed: {program}"
raise ValueError(msg)
if os.environ.get("IN_NIX_SANDBOX"):
return cmd
missing_packages = [

8
pkgs/clan-cli/clan_cli/qemu/qga.py
View File

@@ -58,7 +58,8 @@ class QgaSession:
self.sock.send(status_payload)
result = self.get_response()
if "error" in result and result["error"]["desc"].startswith("PID"):
raise Exception("PID could not be found")
msg = "PID could not be found"
raise Exception(msg)
if result["return"]["exited"]:
break
sleep(0.1)
@@ -75,7 +76,6 @@ class QgaSession:
else base64.b64decode(result["return"]["err-data"]).decode("utf-8")
)
if check and exitcode != 0:
raise Exception(
f"Command on guest failed\nCommand: {cmd}\nExitcode {exitcode}\nStdout: {stdout}\nStderr: {stderr}"
)
msg = f"Command on guest failed\nCommand: {cmd}\nExitcode {exitcode}\nStdout: {stdout}\nStderr: {stderr}"
raise Exception(msg)
return exitcode, stdout, stderr

9
pkgs/clan-cli/clan_cli/qemu/qmp.py
View File

@@ -139,11 +139,14 @@ class QEMUMonitorProtocol:
try:
ret = self.__json_read(only_event=True)
except TimeoutError as e:
raise QMPTimeoutError("Timeout waiting for event") from e
msg = "Timeout waiting for event"
raise QMPTimeoutError(msg) from e
except OSError as e:
raise QMPConnectError("Error while reading from socket") from e
msg = "Error while reading from socket"
raise QMPConnectError(msg) from e
if ret is None:
raise QMPConnectError("Error while reading from socket")
msg = "Error while reading from socket"
raise QMPConnectError(msg)
self.__sock.settimeout(None)
def __enter__(self) -> "QEMUMonitorProtocol":

3
pkgs/clan-cli/clan_cli/secrets/folders.py
View File

@@ -39,7 +39,8 @@ def remove_object(path: Path, name: str) -> list[Path]:
shutil.rmtree(path / name)
paths_to_commit.append(path / name)
except FileNotFoundError as e:
raise ClanError(f"{name} not found in {path}") from e
msg = f"{name} not found in {path}"
raise ClanError(msg) from e
if not os.listdir(path):
os.rmdir(path)
return paths_to_commit

5
pkgs/clan-cli/clan_cli/secrets/groups.py
View File

@@ -120,9 +120,8 @@ def add_member(
user_target = group_folder / name
if user_target.exists():
if not user_target.is_symlink():
raise ClanError(
f"Cannot add user {name}. {user_target} exists but is not a symlink"
)
msg = f"Cannot add user {name}. {user_target} exists but is not a symlink"
raise ClanError(msg)
os.remove(user_target)
user_target.symlink_to(os.path.relpath(source, user_target.parent))
return update_group_keys(flake_dir, group_folder.parent.name)

3
pkgs/clan-cli/clan_cli/secrets/import_sops.py
View File

@@ -23,7 +23,8 @@ def import_sops(args: argparse.Namespace) -> None:
try:
file.read_text()
except OSError as e:
raise ClanError(f"Could not read file {file}: {e}") from e
msg = f"Could not read file {file}: {e}"
raise ClanError(msg) from e
if file_type == ".yaml":
cmd = ["sops"]
if args.input_type:

11
pkgs/clan-cli/clan_cli/secrets/key.py
View File

@@ -23,13 +23,14 @@ def extract_public_key(filepath: Path) -> str:
# Extract and return the public key part after the prefix
return line.strip().split(": ")[1]
except FileNotFoundError as e:
raise ClanError(f"The file at {filepath} was not found.") from e
msg = f"The file at {filepath} was not found."
raise ClanError(msg) from e
except OSError as e:
raise ClanError(
f"An error occurred while extracting the public key: {e}"
) from e
msg = f"An error occurred while extracting the public key: {e}"
raise ClanError(msg) from e
raise ClanError(f"Could not find the public key in the file at {filepath}.")
msg = f"Could not find the public key in the file at {filepath}."
raise ClanError(msg)
def generate_key() -> str:

18
pkgs/clan-cli/clan_cli/secrets/machines.py
View File

@@ -96,7 +96,8 @@ def remove_secret(flake_dir: Path, machine: str, secret: str) -> None:
def list_command(args: argparse.Namespace) -> None:
if args.flake is None:
raise ClanError("Could not find clan flake toplevel directory")
msg = "Could not find clan flake toplevel directory"
raise ClanError(msg)
lst = list_sops_machines(args.flake.path)
if len(lst) > 0:
print("\n".join(lst))
@@ -104,31 +105,36 @@ def list_command(args: argparse.Namespace) -> None:
def add_command(args: argparse.Namespace) -> None:
if args.flake is None:
raise ClanError("Could not find clan flake toplevel directory")
msg = "Could not find clan flake toplevel directory"
raise ClanError(msg)
add_machine(args.flake.path, args.machine, args.key, args.force)
def get_command(args: argparse.Namespace) -> None:
if args.flake is None:
raise ClanError("Could not find clan flake toplevel directory")
msg = "Could not find clan flake toplevel directory"
raise ClanError(msg)
print(get_machine(args.flake.path, args.machine))
def remove_command(args: argparse.Namespace) -> None:
if args.flake is None:
raise ClanError("Could not find clan flake toplevel directory")
msg = "Could not find clan flake toplevel directory"
raise ClanError(msg)
remove_machine(args.flake.path, args.machine)
def add_secret_command(args: argparse.Namespace) -> None:
if args.flake is None:
raise ClanError("Could not find clan flake toplevel directory")
msg = "Could not find clan flake toplevel directory"
raise ClanError(msg)
add_secret(args.flake.path, args.machine, args.secret)
def remove_secret_command(args: argparse.Namespace) -> None:
if args.flake is None:
raise ClanError("Could not find clan flake toplevel directory")
msg = "Could not find clan flake toplevel directory"
raise ClanError(msg)
remove_secret(args.flake.path, args.machine, args.secret)

22
pkgs/clan-cli/clan_cli/secrets/secrets.py
View File

@@ -158,7 +158,8 @@ def encrypt_secret(
def remove_secret(flake_dir: Path, secret: str) -> None:
path = sops_secrets_folder(flake_dir) / secret
if not path.exists():
raise ClanError(f"Secret '{secret}' does not exist")
msg = f"Secret '{secret}' does not exist"
raise ClanError(msg)
shutil.rmtree(path)
commit_files(
[path],
@@ -215,9 +216,8 @@ def allow_member(
user_target = group_folder / name
if user_target.exists():
if not user_target.is_symlink():
raise ClanError(
f"Cannot add user '{name}' to {group_folder.parent.name} secret. {user_target} exists but is not a symlink"
)
msg = f"Cannot add user '{name}' to {group_folder.parent.name} secret. {user_target} exists but is not a symlink"
raise ClanError(msg)
os.remove(user_target)
user_target.symlink_to(os.path.relpath(source, user_target.parent))
@@ -242,9 +242,8 @@ def disallow_member(group_folder: Path, name: str) -> list[Path]:
keys = collect_keys_for_path(group_folder.parent)
if len(keys) < 2:
raise ClanError(
f"Cannot remove {name} from {group_folder.parent.name}. No keys left. Use 'clan secrets remove {name}' to remove the secret."
)
msg = f"Cannot remove {name} from {group_folder.parent.name}. No keys left. Use 'clan secrets remove {name}' to remove the secret."
raise ClanError(msg)
os.remove(target)
if len(os.listdir(group_folder)) == 0:
@@ -295,7 +294,8 @@ def decrypt_secret(flake_dir: Path, secret_path: Path) -> str:
ensure_sops_key(flake_dir)
path = secret_path / "secret"
if not path.exists():
raise ClanError(f"Secret '{secret_path!s}' does not exist")
msg = f"Secret '{secret_path!s}' does not exist"
raise ClanError(msg)
return decrypt_file(path)
@@ -332,9 +332,11 @@ def rename_command(args: argparse.Namespace) -> None:
old_path = sops_secrets_folder(flake_dir) / args.secret
new_path = sops_secrets_folder(flake_dir) / args.new_name
if not old_path.exists():
raise ClanError(f"Secret '{args.secret}' does not exist")
msg = f"Secret '{args.secret}' does not exist"
raise ClanError(msg)
if new_path.exists():
raise ClanError(f"Secret '{args.new_name}' already exists")
msg = f"Secret '{args.new_name}' already exists"
raise ClanError(msg)
os.rename(old_path, new_path)
commit_files(
[old_path, new_path],

46
pkgs/clan-cli/clan_cli/secrets/sops.py
View File

@@ -30,9 +30,8 @@ def get_public_key(privkey: str) -> str:
cmd, input=privkey, stdout=subprocess.PIPE, text=True, check=True
)
except subprocess.CalledProcessError as e:
raise ClanError(
"Failed to get public key for age private key. Is the key malformed?"
) from e
msg = "Failed to get public key for age private key. Is the key malformed?"
raise ClanError(msg) from e
return res.stdout.strip()
@@ -49,15 +48,18 @@ def generate_private_key(out_file: Path | None = None) -> tuple[str, str]:
if not line.startswith("#"):
private_key = line
if not pubkey:
raise ClanError("Could not find public key in age-keygen output")
msg = "Could not find public key in age-keygen output"
raise ClanError(msg)
if not private_key:
raise ClanError("Could not find private key in age-keygen output")
msg = "Could not find private key in age-keygen output"
raise ClanError(msg)
if out_file:
out_file.parent.mkdir(parents=True, exist_ok=True)
out_file.write_text(res)
return private_key, pubkey
except subprocess.CalledProcessError as e:
raise ClanError("Failed to generate private sops key") from e
msg = "Failed to generate private sops key"
raise ClanError(msg) from e
def get_user_name(flake_dir: Path, user: str) -> str:
@@ -86,9 +88,8 @@ def ensure_user_or_machine(flake_dir: Path, pub_key: str) -> SopsKey:
key.username = user.name
return key
raise ClanError(
f"Your sops key is not yet added to the repository. Please add it with 'clan secrets users add youruser {pub_key}' (replace youruser with your user name)"
)
msg = f"Your sops key is not yet added to the repository. Please add it with 'clan secrets users add youruser {pub_key}' (replace youruser with your user name)"
raise ClanError(msg)
def default_sops_key_path() -> Path:
@@ -107,9 +108,8 @@ def ensure_sops_key(flake_dir: Path) -> SopsKey:
if path.exists():
return ensure_user_or_machine(flake_dir, get_public_key(path.read_text()))
else:
raise ClanError(
"No sops key found. Please generate one with 'clan secrets key generate'."
)
msg = "No sops key found. Please generate one with 'clan secrets key generate'."
raise ClanError(msg)
@contextmanager
@@ -164,9 +164,10 @@ def encrypt_file(
p = subprocess.run(cmd, check=False)
# returns 200 if the file is changed
if p.returncode != 0 and p.returncode != 200:
raise ClanError(
msg = (
f"Failed to encrypt {secret_path}: sops exited with {p.returncode}"
)
raise ClanError(msg)
return
# hopefully /tmp is written to an in-memory file to avoid leaking secrets
@@ -182,7 +183,8 @@ def encrypt_file(
with open(f.name, "w") as fd:
shutil.copyfileobj(content, fd)
else:
raise ClanError(f"Invalid content type: {type(content)}")
msg = f"Invalid content type: {type(content)}"
raise ClanError(msg)
# we pass an empty manifest to pick up existing configuration of the user
args = ["sops", "--config", str(manifest)]
args.extend(["-i", "--encrypt", str(f.name)])
@@ -228,9 +230,8 @@ def write_key(path: Path, publickey: str, overwrite: bool) -> None:
flags |= os.O_EXCL
fd = os.open(path / "key.json", flags)
except FileExistsError as e:
raise ClanError(
f"{path.name} already exists in {path}. Use --force to overwrite."
) from e
msg = f"{path.name} already exists in {path}. Use --force to overwrite."
raise ClanError(msg) from e
with os.fdopen(fd, "w") as f:
json.dump({"publickey": publickey, "type": "age"}, f, indent=2)
@@ -240,12 +241,13 @@ def read_key(path: Path) -> str:
try:
key = json.load(f)
except json.JSONDecodeError as e:
raise ClanError(f"Failed to decode {path.name}: {e}") from e
msg = f"Failed to decode {path.name}: {e}"
raise ClanError(msg) from e
if key["type"] != "age":
raise ClanError(
f"{path.name} is not an age key but {key['type']}. This is not supported"
)
msg = f"{path.name} is not an age key but {key['type']}. This is not supported"
raise ClanError(msg)
publickey = key.get("publickey")
if not publickey:
raise ClanError(f"{path.name} does not contain a public key")
msg = f"{path.name} does not contain a public key"
raise ClanError(msg)
return publickey

20
pkgs/clan-cli/clan_cli/secrets/types.py
View File

@@ -14,9 +14,8 @@ VALID_USER_NAME = re.compile(r"^[a-z_]([a-z0-9_-]{0,31})?$")
def secret_name_type(arg_value: str) -> str:
if not VALID_SECRET_NAME.match(arg_value):
raise argparse.ArgumentTypeError(
"Invalid character in secret name. Allowed characters are a-z, A-Z, 0-9, ., -, and _"
)
msg = "Invalid character in secret name. Allowed characters are a-z, A-Z, 0-9, ., -, and _"
raise argparse.ArgumentTypeError(msg)
return arg_value
@@ -28,22 +27,19 @@ def public_or_private_age_key_type(arg_value: str) -> str:
if arg_value.startswith("AGE-SECRET-KEY-"):
return get_public_key(arg_value)
if not arg_value.startswith("age1"):
raise ClanError(
f"Please provide an age key starting with age1, got: '{arg_value}'"
)
msg = f"Please provide an age key starting with age1, got: '{arg_value}'"
raise ClanError(msg)
return arg_value
def group_or_user_name_type(what: str) -> Callable[[str], str]:
def name_type(arg_value: str) -> str:
if len(arg_value) > 32:
raise argparse.ArgumentTypeError(
f"{what.capitalize()} name must be less than 32 characters long"
)
msg = f"{what.capitalize()} name must be less than 32 characters long"
raise argparse.ArgumentTypeError(msg)
if not VALID_USER_NAME.match(arg_value):
raise argparse.ArgumentTypeError(
f"Invalid character in {what} name. Allowed characters are a-z, 0-9, -, and _. Must start with a letter or _"
)
msg = f"Invalid character in {what} name. Allowed characters are a-z, 0-9, -, and _. Must start with a letter or _"
raise argparse.ArgumentTypeError(msg)
return arg_value
return name_type

18
pkgs/clan-cli/clan_cli/secrets/users.py
View File

@@ -84,7 +84,8 @@ def remove_secret(flake_dir: Path, user: str, secret: str) -> None:
def list_command(args: argparse.Namespace) -> None:
if args.flake is None:
raise ClanError("Could not find clan flake toplevel directory")
msg = "Could not find clan flake toplevel directory"
raise ClanError(msg)
lst = list_users(args.flake.path)
if len(lst) > 0:
print("\n".join(lst))
@@ -92,31 +93,36 @@ def list_command(args: argparse.Namespace) -> None:
def add_command(args: argparse.Namespace) -> None:
if args.flake is None:
raise ClanError("Could not find clan flake toplevel directory")
msg = "Could not find clan flake toplevel directory"
raise ClanError(msg)
add_user(args.flake.path, args.user, args.key, args.force)
def get_command(args: argparse.Namespace) -> None:
if args.flake is None:
raise ClanError("Could not find clan flake toplevel directory")
msg = "Could not find clan flake toplevel directory"
raise ClanError(msg)
print(get_user(args.flake.path, args.user))
def remove_command(args: argparse.Namespace) -> None:
if args.flake is None:
raise ClanError("Could not find clan flake toplevel directory")
msg = "Could not find clan flake toplevel directory"
raise ClanError(msg)
remove_user(args.flake.path, args.user)
def add_secret_command(args: argparse.Namespace) -> None:
if args.flake is None:
raise ClanError("Could not find clan flake toplevel directory")
msg = "Could not find clan flake toplevel directory"
raise ClanError(msg)
add_secret(args.flake.path, args.user, args.secret)
def remove_secret_command(args: argparse.Namespace) -> None:
if args.flake is None:
raise ClanError("Could not find clan flake toplevel directory")
msg = "Could not find clan flake toplevel directory"
raise ClanError(msg)
remove_secret(args.flake.path, args.user, args.secret)

17
pkgs/clan-cli/clan_cli/ssh/__init__.py
View File

@@ -295,7 +295,8 @@ class Host:
elif stdout == subprocess.PIPE:
stdout_read, stdout_write = stack.enter_context(_pipe())
else:
raise ClanError(f"unsupported value for stdout parameter: {stdout}")
msg = f"unsupported value for stdout parameter: {stdout}"
raise ClanError(msg)
if stderr is None:
stderr_read = None
@@ -303,7 +304,8 @@ class Host:
elif stderr == subprocess.PIPE:
stderr_read, stderr_write = stack.enter_context(_pipe())
else:
raise ClanError(f"unsupported value for stderr parameter: {stderr}")
msg = f"unsupported value for stderr parameter: {stderr}"
raise ClanError(msg)
env = os.environ.copy()
env.update(extra_env)
@@ -355,7 +357,8 @@ class Host:
return subprocess.CompletedProcess(
cmd, ret, stdout=stdout_data, stderr=stderr_data
)
raise RuntimeError("unreachable")
msg = "unreachable"
raise RuntimeError(msg)
def run_local(
self,
@@ -624,9 +627,8 @@ class HostGroup:
)
errors += 1
if errors > 0:
raise ClanError(
f"{errors} hosts failed with an error. Check the logs above"
)
msg = f"{errors} hosts failed with an error. Check the logs above"
raise ClanError(msg)
def _run(
self,
@@ -802,7 +804,8 @@ def parse_deployment_address(
options[k] = v
result = urllib.parse.urlsplit("//" + hostname)
if not result.hostname:
raise Exception(f"Invalid hostname: {hostname}")
msg = f"Invalid hostname: {hostname}"
raise Exception(msg)
hostname = result.hostname
port = result.port
meta = meta.copy()

6
pkgs/clan-cli/clan_cli/state/list.py
View File

@@ -34,8 +34,9 @@ def list_state_folders(machine: str, service: None | str = None) -> None:
proc = run_no_stdout(cmd)
res = proc.stdout.strip()
except ClanCmdError as e:
msg = "Clan might not have meta attributes"
raise ClanError(
"Clan might not have meta attributes",
msg,
location=f"show_clan {uri}",
description="Evaluation failed on clanInternals.meta attribute",
) from e
@@ -45,8 +46,9 @@ def list_state_folders(machine: str, service: None | str = None) -> None:
if state_info := state.get(service):
state = {service: state_info}
else:
msg = f"Service {service} isn't configured for this machine."
raise ClanError(
f"Service {service} isn't configured for this machine.",
msg,
location=f"clan state list {machine} --service {service}",
description=f"The service: {service} needs to be configured for the machine.",
)

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

@@ -195,7 +195,8 @@ def prompt_func(description: str, input_type: str) -> str:
elif input_type == "hidden":
result = getpass(f"Enter the value for {description} (hidden): ")
else:
raise ClanError(f"Unknown input type: {input_type} for prompt {description}")
msg = f"Unknown input type: {input_type} for prompt {description}"
raise ClanError(msg)
log.info("Input received. Processing...")
return result
@@ -226,9 +227,8 @@ def _generate_vars_for_machine(
if generator_name and generator_name not in machine.vars_generators:
generators = list(machine.vars_generators.keys())
raise ClanError(
f"Could not find generator with name: {generator_name}. The following generators are available: {generators}"
)
msg = f"Could not find generator with name: {generator_name}. The following generators are available: {generators}"
raise ClanError(msg)
graph = {
gen_name: set(generator["dependencies"])
@@ -243,9 +243,8 @@ def _generate_vars_for_machine(
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"
)
msg = f"Generator {gen_name} has a dependency on {dep}, which does not exist"
raise ClanError(msg)
# process generators in topological order
sorter = TopologicalSorter(graph)
@@ -280,9 +279,8 @@ def generate_vars(
log.error(f"Failed to generate facts for {machine.name}: {exc}")
errors += [exc]
if len(errors) > 0:
raise ClanError(
f"Failed to generate facts for {len(errors)} hosts. Check the logs above"
) from errors[0]
msg = f"Failed to generate facts for {len(errors)} hosts. Check the logs above"
raise ClanError(msg) from errors[0]
if not was_regenerated:
print("All secrets and facts are already up to date")

5
pkgs/clan-cli/clan_cli/vars/public_modules/in_repo.py
View File

@@ -31,9 +31,8 @@ class FactStore(FactStoreBase):
fact_path.write_bytes(value)
return fact_path
else:
raise ClanError(
f"in_flake fact storage is only supported for local flakes: {self.machine.flake}"
)
msg = f"in_flake fact storage is only supported for local flakes: {self.machine.flake}"
raise ClanError(msg)
def exists(self, generator_name: str, name: str, shared: bool = False) -> bool:
return self._var_path(generator_name, name, shared).exists()

3
pkgs/clan-cli/clan_cli/vars/public_modules/vm.py
View File

@@ -34,4 +34,5 @@ class FactStore(FactStoreBase):
fact_path = self.dir / service / name
if fact_path.exists():
return fact_path.read_bytes()
raise ClanError(f"Fact {name} for service {service} not found")
msg = f"Fact {name} for service {service} not found"
raise ClanError(msg)

3
pkgs/clan-cli/clan_cli/vms/qemu.py
View File

@@ -169,7 +169,8 @@ class QMPWrapper:
def qmp_ctx(self) -> Generator[QEMUMonitorProtocol, None, None]:
rpath = self._qmp_socket.resolve()
if not rpath.exists():
raise ClanError(f"qmp socket {rpath} does not exist. Is the VM running?")
msg = f"qmp socket {rpath} does not exist. Is the VM running?"
raise ClanError(msg)
qmp = QEMUMonitorProtocol(str(rpath))
qmp.connect()
try:

3
pkgs/clan-cli/clan_cli/vms/run.py
View File

@@ -60,7 +60,8 @@ def build_vm(
vm_data["secrets_dir"] = str(secrets_dir)
return vm_data
except json.JSONDecodeError as e:
raise ClanError(f"Failed to parse vm config: {e}") from e
msg = f"Failed to parse vm config: {e}"
raise ClanError(msg) from e
def get_secrets(