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

PLR2004: fix

Browse Source
This commit is contained in:
Jörg Thalheim
2025-08-26 15:55:02 +02:00
parent f26499edb8
commit 4cb17d42e1
20 changed files with 129 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
clanServices/wireguard/ipv6_allocator.py
View File

@@ -12,6 +12,11 @@ import ipaddress
import sys
from pathlib import Path
# Constants for argument count validation
MIN_ARGS_BASE = 4
MIN_ARGS_CONTROLLER = 5
MIN_ARGS_PEER = 5
def hash_string(s: str) -> str:
"""Generate SHA256 hash of string."""
@@ -77,7 +82,7 @@ def generate_peer_suffix(peer_name: str) -> str:
def main() -> None:
if len(sys.argv) < 4:
if len(sys.argv) < MIN_ARGS_BASE:
print(
"Usage: ipv6_allocator.py <output_dir> <instance_name> <controller|peer> <machine_name>",
)
@@ -91,7 +96,7 @@ def main() -> None:
base_network = generate_ula_prefix(instance_name)
if node_type == "controller":
if len(sys.argv) < 5:
if len(sys.argv) < MIN_ARGS_CONTROLLER:
print("Controller name required")
sys.exit(1)
@@ -107,7 +112,7 @@ def main() -> None:
(output_dir / "prefix").write_text(prefix_str)
elif node_type == "peer":
if len(sys.argv) < 5:
if len(sys.argv) < MIN_ARGS_PEER:
print("Peer name required")
sys.exit(1)

12
nixosModules/clanCore/zerotier/generate.py
View File

@@ -16,6 +16,10 @@ from pathlib import Path
from tempfile import TemporaryDirectory
from typing import Any
# Constants
NODE_ID_LENGTH = 10
NETWORK_ID_LENGTH = 16
class ClanError(Exception):
pass
@@ -55,8 +59,8 @@ class Identity:
def node_id(self) -> str:
nid = self.public.split(":")[0]
if len(nid) != 10:
msg = f"node_id must be 10 characters long, got {len(nid)}: {nid}"
if len(nid) != NODE_ID_LENGTH:
msg = f"node_id must be {NODE_ID_LENGTH} characters long, got {len(nid)}: {nid}"
raise ClanError(msg)
return nid
@@ -173,8 +177,8 @@ def create_identity() -> Identity:
def compute_zerotier_ip(network_id: str, identity: Identity) -> ipaddress.IPv6Address:
if len(network_id) != 16:
msg = f"network_id must be 16 characters long, got '{network_id}'"
if len(network_id) != NETWORK_ID_LENGTH:
msg = f"network_id must be {NETWORK_ID_LENGTH} characters long, got '{network_id}'"
raise ClanError(msg)
nwid = int(network_id, 16)
node_id = int(identity.node_id(), 16)

5
nixosModules/clanCore/zerotier/genmoon.py
View File

@@ -6,9 +6,12 @@ import sys
from pathlib import Path
from tempfile import NamedTemporaryFile
# Constants
REQUIRED_ARGS = 4
def main() -> None:
if len(sys.argv) != 4:
if len(sys.argv) != REQUIRED_ARGS:
print("Usage: genmoon.py <moon.json> <endpoint.json> <moons.d>")
sys.exit(1)
moon_json_path = sys.argv[1]

5
pkgs/clan-cli/clan_cli/flash/flash_cmd.py
View File

@@ -14,6 +14,9 @@ from clan_cli.completions import add_dynamic_completer, complete_machines
log = logging.getLogger(__name__)
# Constants for disk validation
EXPECTED_DISK_VALUES = 2
@dataclass
class FlashOptions:
@@ -44,7 +47,7 @@ class AppendDiskAction(argparse.Action):
if not (
isinstance(values, Sequence)
and not isinstance(values, str)
and len(values) == 2
and len(values) == EXPECTED_DISK_VALUES
):
msg = "Two values must be provided for a 'disk'"
raise ValueError(msg)

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

@@ -3,15 +3,18 @@ import re
VALID_HOSTNAME = re.compile(r"^[a-z0-9]([-a-z0-9]*[a-z0-9])?$", re.IGNORECASE)
# Maximum hostname/machine name length as per RFC specifications
MAX_HOSTNAME_LENGTH = 63
def validate_hostname(hostname: str) -> bool:
if len(hostname) > 63:
if len(hostname) > MAX_HOSTNAME_LENGTH:
return False
return VALID_HOSTNAME.match(hostname) is not None
def machine_name_type(arg_value: str) -> str:
if len(arg_value) > 63:
if len(arg_value) > MAX_HOSTNAME_LENGTH:
msg = "Machine name must be less than 63 characters long"
raise argparse.ArgumentTypeError(msg)
if not VALID_HOSTNAME.match(arg_value):

8
pkgs/clan-cli/clan_cli/profiler.py
View File

@@ -10,6 +10,10 @@ from typing import Any
# Ensure you have a logger set up for logging exceptions
log = logging.getLogger(__name__)
# Constants for path trimming and profiler configuration
MAX_PATH_LEVELS = 4
explanation = """
cProfile Output Columns Explanation:
@@ -86,8 +90,8 @@ class ProfilerStore:
def trim_path_to_three_levels(path: str) -> str:
parts = path.split(os.path.sep)
if len(parts) > 4:
return os.path.sep.join(parts[-4:])
if len(parts) > MAX_PATH_LEVELS:
return os.path.sep.join(parts[-MAX_PATH_LEVELS:])
return path

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

@@ -31,6 +31,9 @@ from .types import VALID_SECRET_NAME, secret_name_type
log = logging.getLogger(__name__)
# Minimum number of keys required to keep a secret group
MIN_KEYS_FOR_GROUP_REMOVAL = 2
def list_generators_secrets(generators_path: Path) -> list[Path]:
paths: list[Path] = []
@@ -328,7 +331,7 @@ def disallow_member(
keys = collect_keys_for_path(group_folder.parent)
if len(keys) < 2:
if len(keys) < MIN_KEYS_FOR_GROUP_REMOVAL:
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)
target.unlink()

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

@@ -10,6 +10,9 @@ from .sops import get_public_age_keys
VALID_SECRET_NAME = re.compile(r"^[a-zA-Z0-9._-]+$")
VALID_USER_NAME = re.compile(r"^[a-z_]([a-z0-9_-]{0,31})?$")
# Maximum length for user and group names
MAX_USER_GROUP_NAME_LENGTH = 32
def secret_name_type(arg_value: str) -> str:
if not VALID_SECRET_NAME.match(arg_value):
@@ -45,7 +48,7 @@ def public_or_private_age_key_type(arg_value: str) -> str:
def group_or_user_name_type(what: str) -> Callable[[str], str]:
def name_type(arg_value: str) -> str:
if len(arg_value) > 32:
if len(arg_value) > MAX_USER_GROUP_NAME_LENGTH:
msg = f"{what.capitalize()} name must be less than 32 characters long"
raise argparse.ArgumentTypeError(msg)
if not VALID_USER_NAME.match(arg_value):

12
pkgs/clan-cli/clan_cli/vars/prompt.py
View File

@@ -14,6 +14,12 @@ log = logging.getLogger(__name__)
# This is for simulating user input in tests.
MOCK_PROMPT_RESPONSE: None = None
# ASCII control character constants
CTRL_D_ASCII = 4 # EOF character
CTRL_C_ASCII = 3 # Interrupt character
DEL_ASCII = 127 # Delete character
BACKSPACE_ASCII = 8 # Backspace character
class PromptType(enum.Enum):
LINE = "line"
@@ -80,14 +86,14 @@ def get_multiline_hidden_input() -> str:
char = sys.stdin.read(1)
# Check for Ctrl-D (ASCII value 4 or EOF)
if not char or ord(char) == 4:
if not char or ord(char) == CTRL_D_ASCII:
# Add last line if not empty
if current_line:
lines.append("".join(current_line))
break
# Check for Ctrl-C (KeyboardInterrupt)
if ord(char) == 3:
if ord(char) == CTRL_C_ASCII:
raise KeyboardInterrupt
# Handle Enter key
@@ -98,7 +104,7 @@ def get_multiline_hidden_input() -> str:
sys.stdout.write("\r\n")
sys.stdout.flush()
# Handle backspace
elif ord(char) == 127 or ord(char) == 8:
elif ord(char) == DEL_ASCII or ord(char) == BACKSPACE_ASCII:
if current_line:
current_line.pop()
# Regular character

21
pkgs/clan-cli/clan_lib/api/mdns_discovery.py
View File

@@ -7,6 +7,13 @@ from clan_lib.nix import nix_shell
from . import API
# Avahi output parsing constants
MIN_NEW_SERVICE_PARTS = (
6 # Minimum parts for new service discovery (+;interface;protocol;name;type;domain)
)
MIN_RESOLVED_SERVICE_PARTS = 9 # Minimum parts for resolved service (=;interface;protocol;name;type;domain;host;ip;port)
TXT_RECORD_INDEX = 9 # Index where TXT record appears in resolved service output
@dataclass
class Host:
@@ -40,7 +47,7 @@ def parse_avahi_output(output: str) -> DNSInfo:
parts = line.split(";")
# New service discovered
# print(parts)
if parts[0] == "+" and len(parts) >= 6:
if parts[0] == "+" and len(parts) >= MIN_NEW_SERVICE_PARTS:
interface, protocol, name, type_, domain = parts[1:6]
name = decode_escapes(name)
@@ -58,7 +65,7 @@ def parse_avahi_output(output: str) -> DNSInfo:
)
# Resolved more data for already discovered services
elif parts[0] == "=" and len(parts) >= 9:
elif parts[0] == "=" and len(parts) >= MIN_RESOLVED_SERVICE_PARTS:
interface, protocol, name, type_, domain, host, ip, port = parts[1:9]
name = decode_escapes(name)
@@ -67,8 +74,10 @@ def parse_avahi_output(output: str) -> DNSInfo:
dns_info.services[name].host = decode_escapes(host)
dns_info.services[name].ip = ip
dns_info.services[name].port = port
if len(parts) > 9:
dns_info.services[name].txt = decode_escapes(parts[9])
if len(parts) > TXT_RECORD_INDEX:
dns_info.services[name].txt = decode_escapes(
parts[TXT_RECORD_INDEX]
)
else:
dns_info.services[name] = Host(
interface=parts[1],
@@ -79,7 +88,9 @@ def parse_avahi_output(output: str) -> DNSInfo:
host=decode_escapes(parts[6]),
ip=parts[7],
port=parts[8],
txt=decode_escapes(parts[9]) if len(parts) > 9 else None,
txt=decode_escapes(parts[TXT_RECORD_INDEX])
if len(parts) > TXT_RECORD_INDEX
else None,
)
return dns_info

10
pkgs/clan-cli/clan_lib/api/type_to_jsonschema.py
View File

@@ -22,6 +22,11 @@ from typing import (
from clan_lib.api.serde import dataclass_to_dict
# Annotation constants
TUPLE_KEY_VALUE_PAIR_LENGTH = (
2 # Expected length for tuple annotations like ("key", value)
)
class JSchemaTypeError(Exception):
pass
@@ -63,7 +68,10 @@ def apply_annotations(schema: dict[str, Any], annotations: list[Any]) -> dict[st
if isinstance(annotation, dict):
# Assuming annotation is a dict that can directly apply to the schema
schema.update(annotation)
elif isinstance(annotation, tuple) and len(annotation) == 2:
elif (
isinstance(annotation, tuple)
and len(annotation) == TUPLE_KEY_VALUE_PAIR_LENGTH
):
# Assuming a tuple where first element is a keyword (like 'minLength') and the second is the value
schema[annotation[0]] = annotation[1]
elif isinstance(annotation, str):

9
pkgs/clan-cli/clan_lib/colors/__init__.py
View File

@@ -5,6 +5,9 @@ ANSI16_MARKER = 300
ANSI256_MARKER = 301
DEFAULT_MARKER = 302
# RGB color constants
RGB_MAX_VALUE = 255 # Maximum value for RGB color components (0-255)
class RgbColor(Enum):
"""A subset of CSS colors with RGB values that work well in Dark and Light mode."""
@@ -107,7 +110,11 @@ def color_code(spec: tuple[int, int, int], base: ColorType) -> str:
val = _join(base.value + 8, 5, green)
elif red == DEFAULT_MARKER:
val = _join(base.value + 9)
elif 0 <= red <= 255 and 0 <= green <= 255 and 0 <= blue <= 255:
elif (
0 <= red <= RGB_MAX_VALUE
and 0 <= green <= RGB_MAX_VALUE
and 0 <= blue <= RGB_MAX_VALUE
):
val = _join(base.value + 8, 2, red, green, blue)
else:
msg = f"Invalid color specification: {spec}"

10
pkgs/clan-cli/clan_lib/log_manager/__init__.py
View File

@@ -8,6 +8,10 @@ from pathlib import Path
log = logging.getLogger(__name__)
# Constants for log parsing
EXPECTED_FILENAME_PARTS = 2 # date_second_str, file_op_key
MIN_PATH_PARTS_FOR_LOGGING = 3 # date/[groups...]/func/file
@dataclass(frozen=True)
class LogGroupConfig:
@@ -559,7 +563,7 @@ class LogManager:
# Parse filename to get op_key and time
filename_stem = log_file_path.stem
parts = filename_stem.split("_", 1)
if len(parts) == 2:
if len(parts) == EXPECTED_FILENAME_PARTS:
date_second_str, file_op_key = parts
if file_op_key == op_key:
@@ -571,7 +575,9 @@ class LogManager:
relative_to_base = log_file_path.relative_to(base_dir)
path_parts = relative_to_base.parts
if len(path_parts) >= 3: # date/[groups...]/func/file
if (
len(path_parts) >= MIN_PATH_PARTS_FOR_LOGGING
): # date/[groups...]/func/file
date_day = path_parts[0]
func_name = path_parts[
-2

9
pkgs/clan-cli/clan_lib/persist/util.py
View File

@@ -10,6 +10,9 @@ from clan_lib.errors import ClanError
T = TypeVar("T")
# Priority constants for configuration merging
WRITABLE_PRIORITY_THRESHOLD = 100 # Values below this are not writeable
empty: list[str] = []
@@ -347,7 +350,7 @@ def determine_writeability(
# If priority is less than 100, all children are not writeable
# If the parent passed "non_writeable" earlier, this makes all children not writeable
if (prio is not None and prio < 100) or non_writeable:
if (prio is not None and prio < WRITABLE_PRIORITY_THRESHOLD) or non_writeable:
results["non_writeable"].add(full_key)
if isinstance(value, dict):
determine_writeability(
@@ -369,7 +372,7 @@ def determine_writeability(
raise ClanError(msg)
is_mergeable = False
if prio == 100:
if prio == WRITABLE_PRIORITY_THRESHOLD:
default = defaults.get(key)
if isinstance(default, dict):
is_mergeable = True
@@ -378,7 +381,7 @@ def determine_writeability(
if key_in_correlated:
is_mergeable = True
is_writeable = prio > 100 or is_mergeable
is_writeable = prio > WRITABLE_PRIORITY_THRESHOLD or is_mergeable
# Append the result
if is_writeable:

9
pkgs/clan-cli/clan_lib/ssh/remote.py
View File

@@ -22,6 +22,9 @@ from clan_lib.ssh.host_key import HostKeyCheck, hostkey_to_ssh_opts
from clan_lib.ssh.socks_wrapper import SocksWrapper
from clan_lib.ssh.sudo_askpass_proxy import SudoAskpassProxy
# Constants for URL parsing
EXPECTED_URL_PARTS = 2 # Expected parts when splitting on '?' or '='
if TYPE_CHECKING:
from clan_lib.network.check import ConnectionOptions
@@ -483,7 +486,9 @@ def _parse_ssh_uri(
address = address.removeprefix("ssh://")
parts = address.split("?", maxsplit=1)
endpoint, maybe_options = parts if len(parts) == 2 else (parts[0], "")
endpoint, maybe_options = (
parts if len(parts) == EXPECTED_URL_PARTS else (parts[0], "")
)
parts = endpoint.split("@")
match len(parts):
@@ -506,7 +511,7 @@ def _parse_ssh_uri(
if len(o) == 0:
continue
parts = o.split("=", maxsplit=1)
if len(parts) != 2:
if len(parts) != EXPECTED_URL_PARTS:
msg = (
f"Invalid option in host `{address}`: option `{o}` does not have "
f"a value (i.e. expected something like `name=value`)"

8
pkgs/clan-cli/clan_lib/ssh/upload.py
View File

@@ -6,6 +6,10 @@ from clan_lib.cmd import Log, RunOpts
from clan_lib.errors import ClanError
from clan_lib.ssh.host import Host
# Safety constants for upload paths
MIN_SAFE_DEPTH = 3 # Minimum path depth for safety
MIN_EXCEPTION_DEPTH = 2 # Minimum depth for allowed exceptions
def upload(
host: Host,
@@ -28,11 +32,11 @@ def upload(
depth = len(remote_dest.parts) - 1
# General rule: destination must be at least 3 levels deep for safety.
is_too_shallow = depth < 3
is_too_shallow = depth < MIN_SAFE_DEPTH
# Exceptions: Allow depth 2 if the path starts with /tmp/, /root/, or /etc/.
# This allows destinations like /tmp/mydir or /etc/conf.d, but not /tmp or /etc directly.
is_allowed_exception = depth >= 2 and (
is_allowed_exception = depth >= MIN_EXCEPTION_DEPTH and (
str(remote_dest).startswith("/tmp/") # noqa: S108 - Path validation check
or str(remote_dest).startswith("/root/")
or str(remote_dest).startswith("/etc/")

5
pkgs/clan-cli/docs.py
View File

@@ -6,6 +6,9 @@ from pathlib import Path
from clan_cli.cli import create_parser
# Constants for command line argument validation
EXPECTED_ARGC = 2 # Expected number of command line arguments
hidden_subcommands = ["machine", "b", "f", "m", "se", "st", "va", "net", "network"]
@@ -380,7 +383,7 @@ def build_command_reference() -> None:
def main() -> None:
if len(sys.argv) != 2:
if len(sys.argv) != EXPECTED_ARGC:
print("Usage: python docs.py <command>")
print("Available commands: reference")
sys.exit(1)

7
pkgs/clan-vm-manager/clan_vm_manager/components/trayicon.py
View File

@@ -91,8 +91,11 @@ class Core:
core = Core()
# Constants
GTK_VERSION_4 = 4
### from pynicotine.gtkgui.application import GTK_API_VERSION
GTK_API_VERSION = 4
GTK_API_VERSION = GTK_VERSION_4
## from pynicotine.gtkgui.application import GTK_GUI_FOLDER_PATH
GTK_GUI_FOLDER_PATH = "assets"
@@ -899,7 +902,7 @@ class Win32Implementation(BaseImplementation):
def _load_ico_buffer(self, icon_name, icon_size):
ico_buffer = b""
if GTK_API_VERSION >= 4:
if GTK_API_VERSION >= GTK_VERSION_4:
icon = ICON_THEME.lookup_icon(
icon_name,
fallbacks=None,

8
pkgs/zerotier-members/zerotier-members.py
View File

@@ -8,13 +8,17 @@ from pathlib import Path
ZEROTIER_STATE_DIR = Path("/var/lib/zerotier-one")
# ZeroTier constants
ZEROTIER_NETWORK_ID_LENGTH = 16 # ZeroTier network ID length
HTTP_OK = 200 # HTTP success status code
class ClanError(Exception):
pass
def compute_zerotier_ip(network_id: str, identity: str) -> ipaddress.IPv6Address:
if len(network_id) != 16:
if len(network_id) != ZEROTIER_NETWORK_ID_LENGTH:
msg = f"network_id must be 16 characters long, got {network_id}"
raise ClanError(msg)
try:
@@ -88,7 +92,7 @@ def allow_member(args: argparse.Namespace) -> None:
{"X-ZT1-AUTH": token},
)
resp = conn.getresponse()
if resp.status != 200:
if resp.status != HTTP_OK:
msg = f"the zerotier daemon returned this error: {resp.status} {resp.reason}"
raise ClanError(msg)
print(resp.status, resp.reason)

4
pyproject.toml
View File

@@ -38,7 +38,6 @@ lint.ignore = [
"G001", # logging-string-format
"G004", # logging-f-string
"PLR0911", # too-many-return-statements
"PLR2004", # magic-value-comparison
"PT023", # pytest-incorrect-mark-parentheses-style
"S603", # subprocess-without-shell-equals-true
"S607", # start-process-with-partial-path
@@ -60,7 +59,8 @@ lint.ignore = [
"{test_*,*_test,**/tests/*}.py" = [
"SLF001", # private-member-access
"S101", # assert
"S105" # hardcoded-password-string
"S105", # hardcoded-password-string
"PLR2004" # magic-value-comparison
]
"**/fixtures/*.py" = [
"S101" # assert