Compare commits

...

2 Commits

Author SHA1 Message Date
d9pouces bf56cdfcd2 feat: can specify several users to configure
Signed-off-by: d9pouces <github@19pouces.net>
2026-10-05 20:46:38 +02:00
d9pouces c498d42da0 ci: version bump
Signed-off-by: d9pouces <github@19pouces.net>
2026-10-04 14:55:09 +02:00
21 changed files with 876 additions and 64 deletions
+1 -1
View File
@@ -77,7 +77,7 @@ repos:
rev: 1.8.6 rev: 1.8.6
hooks: hooks:
- id: bandit - id: bandit
args: [-s, 'B101,B404,B405,B506,B703,B308,B603'] args: [-s, 'B101,B105,B404,B405,B506,B703,B308,B603,B604']
exclude: ^(id_tests/.*|npm/node_modules/.*|tools/.*|doc/.*|interdiode/static/vendor/.*) exclude: ^(id_tests/.*|npm/node_modules/.*|tools/.*|doc/.*|interdiode/static/vendor/.*)
- repo: https://github.com/shellcheck-py/shellcheck-py - repo: https://github.com/shellcheck-py/shellcheck-py
rev: v0.11.0.1 rev: v0.11.0.1
+25 -1
View File
@@ -83,6 +83,26 @@ DstOption = Annotated[
help="Destination directory receiving the modified files.", help="Destination directory receiving the modified files.",
), ),
] ]
UserOption = Annotated[
list[str],
typer.Option(
"--user",
help="A user to prepare per-user configuration for (may be repeated).",
),
]
AutoUsersOption = Annotated[
bool,
typer.Option(
"--auto-users",
help=(
"Also prepare per-user configuration for every regular "
"(non-system) user found in <src>/etc/passwd: one with a login "
"shell (not nologin, sync or false) and a UID at or above "
"UID_MIN (from <src>/etc/login.defs, 1000 by default) - root is "
"always included regardless of its UID."
),
),
]
VerboseOption = Annotated[ VerboseOption = Annotated[
bool, typer.Option("--verbose", "-v", help="Enable verbose (debug) logging.") bool, typer.Option("--verbose", "-v", help="Enable verbose (debug) logging.")
] ]
@@ -121,6 +141,8 @@ def apply(
config_url_cacert_content: ConfigUrlCacertContentOption = None, config_url_cacert_content: ConfigUrlCacertContentOption = None,
config_url_insecure: ConfigUrlInsecureOption = False, config_url_insecure: ConfigUrlInsecureOption = False,
src: SrcOption = None, src: SrcOption = None,
user: UserOption = [], # noqa: B006 (Typer needs a concrete default to detect --user as repeatable)
auto_users: AutoUsersOption = False,
verbose: VerboseOption = False, verbose: VerboseOption = False,
) -> None: ) -> None:
"""Apply a configuration to the destination directory.""" """Apply a configuration to the destination directory."""
@@ -129,7 +151,9 @@ def apply(
config, config_env, config_url, config_url_cacert_content, config_url_insecure config, config_env, config_url, config_url_cacert_content, config_url_insecure
) )
dst.mkdir(parents=True, exist_ok=True) dst.mkdir(parents=True, exist_ok=True)
context = Context(config=configuration, dst=dst, src=src) context = Context(
config=configuration, dst=dst, src=src, users=user, auto_users=auto_users
)
for module_cls in all_modules(): for module_cls in all_modules():
module = module_cls() module = module_cls()
if module.is_relevant(context): if module.is_relevant(context):
+7 -2
View File
@@ -239,11 +239,16 @@ class NpmConfig(BaseModel):
class RustRepository(BaseModel): class RustRepository(BaseModel):
"""A cargo registry (see the "source" tables in .cargo/config.toml), """A cargo registry (see the "source" tables in .cargo/config.toml).
served over the sparse HTTP protocol."""
`is_sparse`, if true, marks the registry as served over the sparse
HTTP protocol (its index is then declared as `sparse+<url>`) rather
than a git-based index.
"""
name: str name: str
url: str url: str
is_sparse: bool = False
class RustConfig(BaseModel): class RustConfig(BaseModel):
+54 -1
View File
@@ -7,18 +7,71 @@ filesystem used to detect the existing configuration.
from __future__ import annotations from __future__ import annotations
from collections.abc import Sequence
from pathlib import Path from pathlib import Path
from id_configurator.config import Configuration from id_configurator.config import Configuration
from id_configurator.users import (
DEFAULT_UID_MIN,
PasswdEntry,
parse_login_defs_uid_min,
parse_passwd,
)
class Context: class Context:
def __init__( def __init__(
self, config: Configuration, dst: Path, src: Path | None = None self,
config: Configuration,
dst: Path,
src: Path | None = None,
users: Sequence[str] | None = None,
auto_users: bool = False,
) -> None: ) -> None:
self.config = config self.config = config
self.dst = dst.resolve() self.dst = dst.resolve()
self.src = src.resolve() if src is not None else None self.src = src.resolve() if src is not None else None
self._passwd_entries = self._load_passwd_entries()
self._uid_min = self._load_uid_min()
self.users = self._resolve_users(users or [], auto_users)
def _load_passwd_entries(self) -> dict[str, PasswdEntry]:
content = self.read_src_file("etc/passwd")
if content is None:
return {}
return {entry.name: entry for entry in parse_passwd(content.decode("utf-8"))}
def _load_uid_min(self) -> int:
content = self.read_src_file("etc/login.defs")
if content is None:
return DEFAULT_UID_MIN
uid_min = parse_login_defs_uid_min(content.decode("utf-8"))
return DEFAULT_UID_MIN if uid_min is None else uid_min
def _resolve_users(self, users: Sequence[str], auto_users: bool) -> list[str]:
resolved: list[str] = []
for user in users:
if user not in resolved:
resolved.append(user)
if auto_users:
for name, entry in self._passwd_entries.items():
if name not in resolved and entry.is_regular_user(self._uid_min):
resolved.append(name)
return resolved
def home_dir(self, user: str) -> str:
"""Relative path (no leading `/`) to `user`'s home directory, as used
by modules that write per-user configuration files.
Taken from `<src>`'s `/etc/passwd` entry for `user` when there is
one (the case for a user resolved via `--auto-users`); otherwise
falls back to the usual `/home/<user>` convention (`/root` for
root).
"""
entry = self._passwd_entries.get(user)
if entry is not None and entry.home:
return entry.home.lstrip("/")
return "root" if user == "root" else f"home/{user}"
def dst_path(self, relative: str) -> Path: def dst_path(self, relative: str) -> Path:
"""Resolve a path relative to the destination directory, refusing to escape it.""" """Resolve a path relative to the destination directory, refusing to escape it."""
+14 -9
View File
@@ -9,14 +9,15 @@ Go's own HTTPS client already trusts whatever the `certificates` module
installs into the system CA store, so no separate certificate configuration installs into the system CA store, so no separate certificate configuration
is needed here, unlike npm's. is needed here, unlike npm's.
Written to `$GOROOT/go.env` - not the per-user `go env -w` file (found at Written to `$GOROOT/go.env`, which `$GOROOT` defaults to `/usr/local/go` for
`os.UserConfigDir()/go/env`, e.g. `$HOME/.config/go/env`), which is a essentially every standard Go install (the official Docker image included);
separate, higher-precedence level in Go's own terminology. `$GOROOT` unlike a config dropped in one specific user's home, this applies regardless
defaults to `/usr/local/go` for essentially every standard Go install (the of which user invokes `go`. The go command reads `$GOROOT/go.env` as a
official Docker image included); unlike a config dropped in one specific fallback whenever a setting is not already present in a higher-precedence
user's home, this applies regardless of which user invokes `go`. The go level - in particular the per-user `go env -w` file (found at
command reads `$GOROOT/go.env` as a fallback whenever a setting is not `os.UserConfigDir()/go/env`, e.g. `$HOME/.config/go/env`), which this module
already present in the user's own go/env file. also writes for each configured `--user` (`context.users`), so that one
takes precedence should that user ever run `go env -w` themselves.
""" """
from __future__ import annotations from __future__ import annotations
@@ -35,7 +36,11 @@ _GO_ENV = "usr/local/go/go.env"
def _write_go_env(context: Context, repositories: list[GoRepository]) -> None: def _write_go_env(context: Context, repositories: list[GoRepository]) -> None:
proxy = ",".join(repository.url for repository in repositories) proxy = ",".join(repository.url for repository in repositories)
lines = [f"GOPROXY={proxy}", "GOSUMDB=off"] lines = [f"GOPROXY={proxy}", "GOSUMDB=off"]
context.write_file(_GO_ENV, ("\n".join(lines) + "\n").encode("utf-8")) content = ("\n".join(lines) + "\n").encode("utf-8")
context.write_file(_GO_ENV, content)
for user in context.users:
context.write_file(f"{context.home_dir(user)}/.config/go/env", content)
logger.info("configured go for user %s", user)
logger.info("configured go to use %d module prox(y/ies)", len(repositories)) logger.info("configured go to use %d module prox(y/ies)", len(repositories))
+15 -3
View File
@@ -19,12 +19,17 @@ when not overridden - the same tradeoff the `php` module's Composer config
already accepts. A `<clear />` element is included so the configured already accepts. A `<clear />` element is included so the configured
sources fully replace nuget.org rather than being merged alongside it, sources fully replace nuget.org rather than being merged alongside it,
matching every other package-manager module's behaviour. matching every other package-manager module's behaviour.
The same file is additionally written under each configured `--user`'s own
home directory (`context.users`), since a non-root user invoking `dotnet`
reads its config from there instead.
""" """
from __future__ import annotations from __future__ import annotations
import logging import logging
import xml.etree.ElementTree as ET import xml.etree.ElementTree as ET
from urllib.parse import urlparse
from id_configurator.config import NuGetRepository from id_configurator.config import NuGetRepository
from id_configurator.context import Context from id_configurator.context import Context
@@ -48,10 +53,17 @@ def _write_nuget_config(context: Context, repositories: list[NuGetRepository]) -
package_sources = ET.SubElement(configuration, "packageSources") package_sources = ET.SubElement(configuration, "packageSources")
ET.SubElement(package_sources, "clear") ET.SubElement(package_sources, "clear")
for repository in repositories: for repository in repositories:
ET.SubElement( attributes = {"key": repository.name, "value": repository.url}
package_sources, "add", {"key": repository.name, "value": repository.url} if urlparse(repository.url).scheme.lower() != "https":
attributes["allowInsecureConnections"] = "true"
ET.SubElement(package_sources, "add", attributes)
content = _render_xml(configuration)
context.write_file(_NUGET_CONFIG, content)
for user in context.users:
context.write_file(
f"{context.home_dir(user)}/.nuget/NuGet/NuGet.Config", content
) )
context.write_file(_NUGET_CONFIG, _render_xml(configuration)) logger.info("configured nuget for user %s", user)
logger.info("configured nuget to use %d package source(s)", len(repositories)) logger.info("configured nuget to use %d package source(s)", len(repositories))
+27 -7
View File
@@ -3,12 +3,20 @@
Composer has no "replace the default registry" switch either: Packagist Composer has no "replace the default registry" switch either: Packagist
stays enabled unless a repository is explicitly declared under the name stays enabled unless a repository is explicitly declared under the name
"packagist.org" and disabled. This just makes the configured repositories "packagist.org" and disabled. If any configured repository's URL isn't
HTTPS, Composer refuses to use it unless `secure-http` is turned off in
`config`; since that effectively means this is a private, non-internet
setup, Packagist - unreachable in such a setup anyway - is disabled at the
same time, exactly as if a `{"packagist.org": false}` repository entry had
been declared. This just makes the configured repositories
available system-wide, via Composer's global configuration file: available system-wide, via Composer's global configuration file:
`/root/.composer/config.json` (`COMPOSER_HOME/config.json`; this tool `/root/.composer/config.json` (`COMPOSER_HOME/config.json`; this tool
targets Docker images, which are built and run as root, so `COMPOSER_HOME` targets Docker images, which are built and run as root, so `COMPOSER_HOME`
is assumed to default to `/root/.composer`, exactly like a fresh Composer is assumed to default to `/root/.composer`, exactly like a fresh Composer
install does when not overridden). install does when not overridden). The same file is additionally written
under each configured `--user`'s own home directory (`context.users`),
since a non-root user invoking `composer` reads its config from there
instead.
Composer's own HTTPS client (PHP's stream wrapper, generally backed by Composer's own HTTPS client (PHP's stream wrapper, generally backed by
curl) already trusts whatever the `certificates` module installs into the curl) already trusts whatever the `certificates` module installs into the
@@ -31,6 +39,7 @@ from __future__ import annotations
import json import json
import logging import logging
from urllib.parse import urlparse
from id_configurator.config import PhpRepository from id_configurator.config import PhpRepository
from id_configurator.context import Context from id_configurator.context import Context
@@ -43,16 +52,27 @@ _CA_BUNDLE = "/etc/ssl/certs/ca-certificates.crt"
_OPENSSL_INI = "usr/local/etc/php/conf.d/id-configurator-openssl.ini" _OPENSSL_INI = "usr/local/etc/php/conf.d/id-configurator-openssl.ini"
def _is_https(url: str) -> bool:
return urlparse(url).scheme.lower() == "https"
def _write_composer_config(context: Context, repositories: list[PhpRepository]) -> None: def _write_composer_config(context: Context, repositories: list[PhpRepository]) -> None:
content = { composer_repositories: list[dict[str, str | bool]] = [
"repositories": [
{"name": repository.name, "type": "composer", "url": repository.url} {"name": repository.name, "type": "composer", "url": repository.url}
for repository in repositories for repository in repositories
] ]
} content: dict[str, object] = {"repositories": composer_repositories}
context.write_file( composer_repositories.append({"packagist.org": False})
_COMPOSER_CONFIG, (json.dumps(content, indent=4) + "\n").encode("utf-8") if any(not _is_https(repository.url) for repository in repositories):
content["config"] = {"secure-http": False}
logger.info(
"disabled secure-http and packagist.org: non-HTTPS repository configured"
) )
rendered = (json.dumps(content, indent=4) + "\n").encode("utf-8")
context.write_file(_COMPOSER_CONFIG, rendered)
for user in context.users:
context.write_file(f"{context.home_dir(user)}/.composer/config.json", rendered)
logger.info("configured composer for user %s", user)
logger.info( logger.info(
"configured composer to use %d additional repositories", len(repositories) "configured composer to use %d additional repositories", len(repositories)
) )
+10 -4
View File
@@ -7,9 +7,11 @@ certificates installed by the certificates module.
- poetry: `/root/.config/pypoetry/config.toml` (`repositories.*`/`certificates.*`). - poetry: `/root/.config/pypoetry/config.toml` (`repositories.*`/`certificates.*`).
Poetry has no global "replace PyPI" switch: a project's `pyproject.toml` Poetry has no global "replace PyPI" switch: a project's `pyproject.toml`
still has to reference these repositories as sources. This only makes the still has to reference these repositories as sources. This only makes the
repository and its certificate available system-wide. It also assumes the repository and its certificate available system-wide. Since poetry only
container runs as root, since poetry only reads its config from the reads its config from the current user's home directory, this assumes the
current user's home directory. container runs as root; the same file is additionally written under each
configured `--user`'s own home directory (`context.users`), for the
benefit of a non-root user invoking poetry.
- easy_install: `distutils.cfg` (`[easy_install] index_url`), written next - easy_install: `distutils.cfg` (`[easy_install] index_url`), written next
to every Python installation detected under `<src>` (there is no location to every Python installation detected under `<src>` (there is no location
independent of a specific Python prefix). independent of a specific Python prefix).
@@ -73,9 +75,13 @@ def _write_poetry_config(
lines.append(f"[certificates.{repository.name}]") lines.append(f"[certificates.{repository.name}]")
lines.append(f"cert = {_toml_string(_CA_BUNDLE)}") lines.append(f"cert = {_toml_string(_CA_BUNDLE)}")
lines.append("") lines.append("")
content = ("\n".join(lines).rstrip() + "\n").encode("utf-8")
context.write_file(_POETRY_CONFIG, content)
for user in context.users:
context.write_file( context.write_file(
_POETRY_CONFIG, ("\n".join(lines).rstrip() + "\n").encode("utf-8") f"{context.home_dir(user)}/.config/pypoetry/config.toml", content
) )
logger.info("configured poetry for user %s", user)
logger.info("registered %d poetry repositories", len(repositories)) logger.info("registered %d poetry repositories", len(repositories))
+40 -8
View File
@@ -1,8 +1,10 @@
"""Configure cargo to use the configured RustRepository entries: the first """Configure cargo to use the configured RustRepository entries: the first
one replaces crates.io entirely (via `[source.crates-io] replace-with`), and one replaces crates.io entirely (via `[source.crates-io] replace-with`), and
any further ones are registered as additional named registries (usable via any further ones are registered as additional named registries (usable via
`cargo add --registry <name>` or a dependency's `registry = "<name>"`), all `cargo add --registry <name>` or a dependency's `registry = "<name>"`). A
using the sparse (plain HTTPS) protocol rather than a git-based index. repository's index is declared as `sparse+<url>` - the sparse (plain HTTPS)
protocol - only when its `is_sparse` option is set; otherwise `<url>` is used
as-is, for a git-based index.
Cargo's own HTTPS client already trusts whatever the `certificates` module Cargo's own HTTPS client already trusts whatever the `certificates` module
installs into the system CA store, so no separate certificate configuration installs into the system CA store, so no separate certificate configuration
@@ -17,11 +19,17 @@ directory to `/`, consulting `/.cargo/config.toml` there, *before* falling
back to `$CARGO_HOME/config.toml` - so this is the one genuinely global back to `$CARGO_HOME/config.toml` - so this is the one genuinely global
location: found regardless of which user, working directory or `CARGO_HOME` location: found regardless of which user, working directory or `CARGO_HOME`
is in effect, unlike a config dropped in one specific user's home. is in effect, unlike a config dropped in one specific user's home.
For each configured `--user` (`context.users`), the same config is
additionally written at that user's own `$HOME/.cargo/config.toml` - the
conventional, un-overridden `CARGO_HOME` location for that specific user,
which this module otherwise has no visibility into.
""" """
from __future__ import annotations from __future__ import annotations
import logging import logging
import re
from id_configurator.config import RustRepository from id_configurator.config import RustRepository
from id_configurator.context import Context from id_configurator.context import Context
@@ -30,24 +38,48 @@ from id_configurator.modules.base import Module, register
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
_CARGO_CONFIG = ".cargo/config.toml" _CARGO_CONFIG = ".cargo/config.toml"
_REPOSITORY_NAME_FORBIDDEN_CHARS = re.compile(r"[^A-Za-z0-9_-]")
def _sanitize_repository_name(name: str) -> str:
"""Replace every character of `name` that isn't a plain identifier
character with `_`.
`name` is written verbatim both as a TOML table key (`[source.name]`,
`[registries.name]`) and inside a quoted string (`replace-with =
"name"`): an unrestricted name could break out of the table header or
string, or (via a literal `.`) silently create extra, nested tables
instead of naming the one table it is meant to.
"""
return _REPOSITORY_NAME_FORBIDDEN_CHARS.sub("_", name)
def _index_url(repository: RustRepository) -> str:
return f"sparse+{repository.url}" if repository.is_sparse else repository.url
def _write_cargo_config(context: Context, repositories: list[RustRepository]) -> None: def _write_cargo_config(context: Context, repositories: list[RustRepository]) -> None:
primary = repositories[0] primary = repositories[0]
primary_name = _sanitize_repository_name(primary.name)
lines = [ lines = [
"[source.crates-io]", "[source.crates-io]",
f'replace-with = "{primary.name}"', f'replace-with = "{primary_name}"',
"", "",
f"[source.{primary.name}]", f"[source.{primary_name}]",
f'registry = "sparse+{primary.url}"', f'registry = "{_index_url(primary)}"',
] ]
for repository in repositories[1:]: for repository in repositories[1:]:
name = _sanitize_repository_name(repository.name)
lines += [ lines += [
"", "",
f"[registries.{repository.name}]", f"[registries.{name}]",
f'index = "sparse+{repository.url}"', f'index = "{_index_url(repository)}"',
] ]
context.write_file(_CARGO_CONFIG, ("\n".join(lines) + "\n").encode("utf-8")) content = ("\n".join(lines) + "\n").encode("utf-8")
context.write_file(_CARGO_CONFIG, content)
for user in context.users:
context.write_file(f"{context.home_dir(user)}/.cargo/config.toml", content)
logger.info("configured cargo for user %s", user)
logger.info("configured cargo to use %s", primary.url) logger.info("configured cargo to use %s", primary.url)
+19 -4
View File
@@ -32,6 +32,7 @@ from __future__ import annotations
import configparser import configparser
import fnmatch import fnmatch
import logging import logging
import re
import shutil import shutil
import subprocess import subprocess
import tempfile import tempfile
@@ -46,6 +47,19 @@ logger = logging.getLogger(__name__)
_REPOS_DIR = "etc/yum.repos.d" _REPOS_DIR = "etc/yum.repos.d"
_RPM_GPG_DIR = "etc/pki/rpm-gpg" _RPM_GPG_DIR = "etc/pki/rpm-gpg"
_REPOSITORY_NAME_FORBIDDEN_CHARS = re.compile(r"[^A-Za-z0-9._-]")
def _sanitize_repository_name(name: str) -> str:
"""Replace every character of `name` that isn't a plain identifier
character with `_`.
`name` is written verbatim as the `.repo` file's INI section header
(`[name]`): an unrestricted name (containing `]`, a newline, ...) could
otherwise inject extra, attacker-controlled sections into the file
instead of just naming the one section it is meant to.
"""
return _REPOSITORY_NAME_FORBIDDEN_CHARS.sub("_", name)
def _disable_repo_content(content: str, patterns: Sequence[str]) -> str: def _disable_repo_content(content: str, patterns: Sequence[str]) -> str:
@@ -91,9 +105,10 @@ def _disable_existing_repos(context: Context, patterns: Sequence[str]) -> None:
def _write_repo_file(context: Context, repository: YumRepository) -> None: def _write_repo_file(context: Context, repository: YumRepository) -> None:
name = _sanitize_repository_name(repository.name)
lines = [ lines = [
f"[{repository.name}]", f"[{name}]",
f"name={repository.name}", f"name={name}",
f"baseurl={repository.index_url}", f"baseurl={repository.index_url}",
"enabled=1", "enabled=1",
] ]
@@ -103,10 +118,10 @@ def _write_repo_file(context: Context, repository: YumRepository) -> None:
else: else:
lines.append("gpgcheck=0") lines.append("gpgcheck=0")
context.write_file( context.write_file(
f"{_REPOS_DIR}/{sanitize_filename(repository.name)}.repo", f"{_REPOS_DIR}/{name}.repo",
("\n".join(lines) + "\n").encode("utf-8"), ("\n".join(lines) + "\n").encode("utf-8"),
) )
logger.info("added yum/dnf repository %s", repository.name) logger.info("added yum/dnf repository %s", name)
def _validate_gpg_key(name: str, content: bytes) -> None: def _validate_gpg_key(name: str, content: bytes) -> None:
+78
View File
@@ -0,0 +1,78 @@
"""Parse `/etc/passwd` entries for the `--auto-users` CLI option.
A passwd(5) line is `name:passwd:uid:gid:gecos:home:shell`. `--auto-users`
expands to every entry that looks like a regular (non-system) account - see
`PasswdEntry.is_regular_user`.
"""
from __future__ import annotations
from dataclasses import dataclass
DEFAULT_UID_MIN = 1000
_DISABLED_SHELL_NAMES = {"nologin", "sync", "false"}
@dataclass(frozen=True)
class PasswdEntry:
name: str
uid: int
home: str
shell: str
@property
def has_login_shell(self) -> bool:
return self.shell.rsplit("/", 1)[-1] not in _DISABLED_SHELL_NAMES
def is_regular_user(self, uid_min: int = DEFAULT_UID_MIN) -> bool:
"""True unless `self` looks like a system account.
A login-disabling shell (`has_login_shell` False) only catches
service accounts that follow that convention; it misses others
(and could wrongly flag a merely locked human account). The
authoritative signal `useradd`/`adduser`/`systemd-sysusers` use
instead is the UID: system accounts get one below `uid_min` (the
`UID_MIN` from login.defs(5), 1000 by default). Both checks apply
together here - except for `root`, whose UID 0 is itself a system
convention rather than a sign it shouldn't be configured: it's
exempted from the UID check (though not from the shell one, so a
deliberately locked root account is still excluded).
"""
if not self.has_login_shell:
return False
return self.name == "root" or self.uid >= uid_min
def parse_passwd(content: str) -> list[PasswdEntry]:
entries = []
for line in content.splitlines():
if not line or line.startswith("#"):
continue
fields = line.split(":")
if len(fields) < 7:
continue
try:
uid = int(fields[2])
except ValueError:
continue
entries.append(
PasswdEntry(name=fields[0], uid=uid, home=fields[5], shell=fields[6]) # noqa: B604
)
return entries
def parse_login_defs_uid_min(content: str) -> int | None:
"""Extract `UID_MIN` from `/etc/login.defs` content (login.defs(5)), or
`None` if it isn't set there."""
for line in content.splitlines():
stripped = line.strip()
if not stripped or stripped.startswith("#"):
continue
parts = stripped.split()
if len(parts) >= 2 and parts[0] == "UID_MIN":
try:
return int(parts[1])
except ValueError:
return None
return None
+1 -1
View File
@@ -44,7 +44,7 @@ maintainers = [
name = "id-configurator" name = "id-configurator"
readme = "README.md" readme = "README.md"
requires-python = ">=3.13" requires-python = ">=3.13"
version = "2026.10.3" version = "2026.10.4"
[project.scripts] [project.scripts]
id-configurator = "id_configurator:main" id-configurator = "id_configurator:main"
+66
View File
@@ -15,6 +15,12 @@ CONFIG_WITH_CERT = {
**MINIMAL_CONFIG, **MINIMAL_CONFIG,
"configurations": {"certificates": [{"name": "my-ca", "content": "fake"}]}, "configurations": {"certificates": [{"name": "my-ca", "content": "fake"}]},
} }
CONFIG_WITH_GO_REPO = {
**MINIMAL_CONFIG,
"configurations": {
"go": {"repositories": [{"name": "demo", "url": "https://example.invalid/"}]}
},
}
def write_config(tmp_path, data) -> str: def write_config(tmp_path, data) -> str:
@@ -129,6 +135,66 @@ def test_apply_runs_relevant_module(tmp_path) -> None:
assert (dst / "etc/ssl/certs/ca-certificates.crt").read_bytes() == b"fake\n" assert (dst / "etc/ssl/certs/ca-certificates.crt").read_bytes() == b"fake\n"
def test_apply_with_user_option_writes_per_user_config(tmp_path) -> None:
config_path = write_config(tmp_path, CONFIG_WITH_GO_REPO)
dst = tmp_path / "dst"
result = runner.invoke(
cli.app,
[
"apply",
"--config",
config_path,
"--dst",
str(dst),
"--user",
"alice",
"--user",
"root",
],
)
assert result.exit_code == 0
assert (dst / "home/alice/.config/go/env").read_text(encoding="utf-8") == (
"GOPROXY=https://example.invalid/\nGOSUMDB=off\n"
)
assert (dst / "root/.config/go/env").read_text(encoding="utf-8") == (
"GOPROXY=https://example.invalid/\nGOSUMDB=off\n"
)
def test_apply_with_auto_users_reads_passwd_from_src(tmp_path) -> None:
config_path = write_config(tmp_path, CONFIG_WITH_GO_REPO)
src = tmp_path / "src"
(src / "etc").mkdir(parents=True)
(src / "etc/passwd").write_text(
"root:x:0:0:root:/root:/bin/bash\n"
"alice:x:1000:1000:Alice:/home/alice:/bin/bash\n"
"daemon:x:1:1:daemon:/usr/sbin:/usr/sbin/nologin\n",
encoding="utf-8",
)
dst = tmp_path / "dst"
result = runner.invoke(
cli.app,
[
"apply",
"--config",
config_path,
"--dst",
str(dst),
"--src",
str(src),
"--auto-users",
],
)
assert result.exit_code == 0
# root is always included despite its uid 0; daemon (nologin shell) is
# a system account, excluded.
assert (dst / "root/.config/go/env").exists()
assert (dst / "home/alice/.config/go/env").exists()
assert not (dst / "usr/sbin/.config/go/env").exists()
def test_main_default_name_calls_app(monkeypatch) -> None: def test_main_default_name_calls_app(monkeypatch) -> None:
mock_app = MagicMock() mock_app = MagicMock()
monkeypatch.setattr(cli, "app", mock_app) monkeypatch.setattr(cli, "app", mock_app)
+132 -2
View File
@@ -8,14 +8,16 @@ CONFIG = Configuration.model_validate(
) )
def make_context(tmp_path, with_src: bool = True) -> Context: def make_context(
tmp_path, with_src: bool = True, users=None, auto_users: bool = False
) -> Context:
dst = tmp_path / "dst" dst = tmp_path / "dst"
dst.mkdir() dst.mkdir()
src = None src = None
if with_src: if with_src:
src = tmp_path / "src" src = tmp_path / "src"
src.mkdir() src.mkdir()
return Context(config=CONFIG, dst=dst, src=src) return Context(config=CONFIG, dst=dst, src=src, users=users, auto_users=auto_users)
def test_dst_path_within_destination(tmp_path) -> None: def test_dst_path_within_destination(tmp_path) -> None:
@@ -97,3 +99,131 @@ def test_glob_src_matches_pattern(tmp_path) -> None:
(context.src / "usr/lib/python-legacy").mkdir(parents=True) (context.src / "usr/lib/python-legacy").mkdir(parents=True)
result = context.glob_src("usr/lib/python3.*") result = context.glob_src("usr/lib/python3.*")
assert [entry.name for entry in result] == ["python3.11", "python3.12"] assert [entry.name for entry in result] == ["python3.11", "python3.12"]
def test_users_defaults_to_empty_list(tmp_path) -> None:
context = make_context(tmp_path, with_src=False)
assert context.users == []
def test_users_keeps_explicit_names_in_order_without_duplicates(tmp_path) -> None:
context = make_context(tmp_path, with_src=False, users=["alice", "bob", "alice"])
assert context.users == ["alice", "bob"]
def test_auto_users_expands_from_passwd_excluding_disabled_shells(tmp_path) -> None:
dst = tmp_path / "dst"
dst.mkdir()
src = tmp_path / "src"
(src / "etc").mkdir(parents=True)
(src / "etc/passwd").write_text(
"root:x:0:0:root:/root:/bin/bash\n"
"alice:x:1000:1000:Alice:/home/alice:/bin/bash\n"
"daemon:x:1:1:daemon:/usr/sbin:/usr/sbin/nologin\n",
encoding="utf-8",
)
context = Context(config=CONFIG, dst=dst, src=src, auto_users=True)
# root is always included despite its uid 0; daemon (nologin shell) is
# excluded as a system account.
assert context.users == ["root", "alice"]
def test_auto_users_excludes_root_with_disabled_shell(tmp_path) -> None:
dst = tmp_path / "dst"
dst.mkdir()
src = tmp_path / "src"
(src / "etc").mkdir(parents=True)
(src / "etc/passwd").write_text(
"root:x:0:0:root:/root:/usr/sbin/nologin\n"
"alice:x:1000:1000:Alice:/home/alice:/bin/bash\n",
encoding="utf-8",
)
context = Context(config=CONFIG, dst=dst, src=src, auto_users=True)
assert context.users == ["alice"]
def test_auto_users_excludes_uid_below_default_uid_min(tmp_path) -> None:
dst = tmp_path / "dst"
dst.mkdir()
src = tmp_path / "src"
(src / "etc").mkdir(parents=True)
(src / "etc/passwd").write_text(
"svc:x:999:999:service account:/home/svc:/bin/bash\n"
"alice:x:1000:1000:Alice:/home/alice:/bin/bash\n",
encoding="utf-8",
)
context = Context(config=CONFIG, dst=dst, src=src, auto_users=True)
assert context.users == ["alice"]
def test_auto_users_honors_uid_min_from_login_defs(tmp_path) -> None:
dst = tmp_path / "dst"
dst.mkdir()
src = tmp_path / "src"
(src / "etc").mkdir(parents=True)
(src / "etc/passwd").write_text(
"svc:x:999:999:service account:/home/svc:/bin/bash\n"
"alice:x:1000:1000:Alice:/home/alice:/bin/bash\n",
encoding="utf-8",
)
(src / "etc/login.defs").write_text("UID_MIN\t500\n", encoding="utf-8")
context = Context(config=CONFIG, dst=dst, src=src, auto_users=True)
assert context.users == ["svc", "alice"]
def test_auto_users_combines_with_explicit_names_without_duplicates(tmp_path) -> None:
dst = tmp_path / "dst"
dst.mkdir()
src = tmp_path / "src"
src.mkdir()
(src / "etc").mkdir()
(src / "etc/passwd").write_text(
"root:x:0:0:root:/root:/bin/bash\nalice:x:1000:1000:Alice:/home/alice:/bin/bash\n",
encoding="utf-8",
)
context = Context(
config=CONFIG, dst=dst, src=src, users=["bob", "alice"], auto_users=True
)
# explicit names come first, in order; auto-detected names are appended,
# skipping "alice" since it's already present.
assert context.users == ["bob", "alice", "root"]
def test_auto_users_false_ignores_passwd(tmp_path) -> None:
dst = tmp_path / "dst"
dst.mkdir()
src = tmp_path / "src"
(src / "etc").mkdir(parents=True)
(src / "etc/passwd").write_text(
"alice:x:1000:1000:Alice:/home/alice:/bin/bash\n", encoding="utf-8"
)
context = Context(config=CONFIG, dst=dst, src=src)
assert context.users == []
def test_auto_users_without_src_is_empty(tmp_path) -> None:
context = make_context(tmp_path, with_src=False, auto_users=True)
assert context.users == []
def test_home_dir_for_root(tmp_path) -> None:
context = make_context(tmp_path, with_src=False)
assert context.home_dir("root") == "root"
def test_home_dir_falls_back_to_convention(tmp_path) -> None:
context = make_context(tmp_path, with_src=False)
assert context.home_dir("alice") == "home/alice"
def test_home_dir_uses_passwd_entry_when_available(tmp_path) -> None:
dst = tmp_path / "dst"
dst.mkdir()
src = tmp_path / "src"
src.mkdir()
(src / "etc").mkdir()
(src / "etc/passwd").write_text(
"alice:x:1000:1000:Alice:/srv/alice:/bin/bash\n", encoding="utf-8"
)
context = Context(config=CONFIG, dst=dst, src=src)
assert context.home_dir("alice") == "srv/alice"
+12
View File
@@ -55,6 +55,18 @@ def test_apply_writes_goproxy(tmp_path) -> None:
assert go_env == "GOPROXY=https://example.invalid/\nGOSUMDB=off\n" assert go_env == "GOPROXY=https://example.invalid/\nGOSUMDB=off\n"
def test_apply_writes_go_env_for_each_configured_user(tmp_path) -> None:
dst = tmp_path / "dst"
context = Context(
config=make_config(CONFIG_WITH_REPO), dst=dst, src=None, users=["alice"]
)
GoModule().apply(context)
go_env = (dst / "home/alice/.config/go/env").read_text(encoding="utf-8")
assert go_env == "GOPROXY=https://example.invalid/\nGOSUMDB=off\n"
def test_apply_joins_multiple_repositories_with_comma(tmp_path) -> None: def test_apply_joins_multiple_repositories_with_comma(tmp_path) -> None:
dst = tmp_path / "dst" dst = tmp_path / "dst"
context = Context(config=make_config(CONFIG_WITH_TWO_REPOS), dst=dst, src=None) context = Context(config=make_config(CONFIG_WITH_TWO_REPOS), dst=dst, src=None)
+42
View File
@@ -63,6 +63,48 @@ def test_apply_writes_nuget_config(tmp_path) -> None:
) )
def test_apply_marks_non_https_source_as_insecure(tmp_path) -> None:
data = {
"uuid": "12345678-1234-5678-1234-567812345678",
"organisation": "InterDiode",
"configurations": {
"nuget": {
"repositories": [{"name": "demo", "url": "http://example.invalid/"}]
}
},
}
dst = tmp_path / "dst"
context = Context(config=make_config(data), dst=dst, src=None)
NuGetModule().apply(context)
content = (dst / "root/.nuget/NuGet/NuGet.Config").read_text(encoding="utf-8")
assert content == (
'<?xml version="1.0" encoding="utf-8"?>\n'
"<configuration>\n"
" <packageSources>\n"
" <clear />\n"
' <add key="demo" value="http://example.invalid/" allowInsecureConnections="true" />\n'
" </packageSources>\n"
"</configuration>\n"
)
def test_apply_writes_config_for_each_configured_user(tmp_path) -> None:
dst = tmp_path / "dst"
context = Context(
config=make_config(CONFIG_WITH_REPO), dst=dst, src=None, users=["alice", "root"]
)
NuGetModule().apply(context)
root_content = (dst / "root/.nuget/NuGet/NuGet.Config").read_text(encoding="utf-8")
alice_content = (dst / "home/alice/.nuget/NuGet/NuGet.Config").read_text(
encoding="utf-8"
)
assert alice_content == root_content
def test_apply_writes_multiple_sources_in_order(tmp_path) -> None: def test_apply_writes_multiple_sources_in_order(tmp_path) -> None:
dst = tmp_path / "dst" dst = tmp_path / "dst"
context = Context(config=make_config(CONFIG_WITH_TWO_REPOS), dst=dst, src=None) context = Context(config=make_config(CONFIG_WITH_TWO_REPOS), dst=dst, src=None)
+66 -2
View File
@@ -80,7 +80,8 @@ def test_apply_writes_global_composer_config(tmp_path) -> None:
"name": "demo", "name": "demo",
"type": "composer", "type": "composer",
"url": "https://example.invalid/composer", "url": "https://example.invalid/composer",
} },
{"packagist.org": False},
] ]
} }
@@ -93,7 +94,70 @@ def test_apply_writes_multiple_repositories_in_order(tmp_path) -> None:
config_path = dst / "root/.composer/config.json" config_path = dst / "root/.composer/config.json"
content = json.loads(config_path.read_text(encoding="utf-8")) content = json.loads(config_path.read_text(encoding="utf-8"))
assert [repo["name"] for repo in content["repositories"]] == ["demo", "other"] names = [repo["name"] for repo in content["repositories"] if "name" in repo]
assert names == ["demo", "other"]
assert {"packagist.org": False} in content["repositories"]
def test_apply_disables_secure_http_and_packagist_when_repo_not_https(
tmp_path,
) -> None:
data = {
"uuid": "12345678-1234-5678-1234-567812345678",
"organisation": "InterDiode",
"configurations": {
"php": {
"repositories": [
{"name": "demo", "url": "https://example.invalid/composer"},
{"name": "insecure", "url": "http://other.invalid/composer"},
]
}
},
}
dst = tmp_path / "dst"
context = Context(config=make_config(data), dst=dst, src=None)
PhpModule().apply(context)
config_path = dst / "root/.composer/config.json"
content = json.loads(config_path.read_text(encoding="utf-8"))
assert content["config"] == {"secure-http": False}
assert {"packagist.org": False} in content["repositories"]
def test_apply_does_not_disable_secure_http_when_all_repos_https(tmp_path) -> None:
dst = tmp_path / "dst"
context = Context(config=make_config(CONFIG_WITH_REPO), dst=dst, src=None)
PhpModule().apply(context)
config_path = dst / "root/.composer/config.json"
content = json.loads(config_path.read_text(encoding="utf-8"))
assert "config" not in content
def test_apply_always_disables_packagist(tmp_path) -> None:
dst = tmp_path / "dst"
context = Context(config=make_config(CONFIG_WITH_REPO), dst=dst, src=None)
PhpModule().apply(context)
config_path = dst / "root/.composer/config.json"
content = json.loads(config_path.read_text(encoding="utf-8"))
assert {"packagist.org": False} in content["repositories"]
def test_apply_writes_composer_config_for_each_configured_user(tmp_path) -> None:
dst = tmp_path / "dst"
context = Context(
config=make_config(CONFIG_WITH_REPO), dst=dst, src=None, users=["alice"]
)
PhpModule().apply(context)
root_content = (dst / "root/.composer/config.json").read_bytes()
alice_content = (dst / "home/alice/.composer/config.json").read_bytes()
assert alice_content == root_content
def test_apply_without_certificates_does_not_write_openssl_ini(tmp_path) -> None: def test_apply_without_certificates_does_not_write_openssl_ini(tmp_path) -> None:
+16
View File
@@ -105,6 +105,22 @@ def test_apply_multiple_repositories(tmp_path) -> None:
) )
def test_apply_writes_poetry_config_for_each_configured_user(tmp_path) -> None:
config = Configuration.model_validate(ONE_REPO_CONFIG)
dst = tmp_path / "dst"
context = Context(config=config, dst=dst, src=None, users=["alice"])
PythonModule().apply(context)
root_content = (dst / "root/.config/pypoetry/config.toml").read_text(
encoding="utf-8"
)
alice_content = (dst / "home/alice/.config/pypoetry/config.toml").read_text(
encoding="utf-8"
)
assert alice_content == root_content
def test_apply_writes_distutils_cfg_for_detected_python_prefixes(tmp_path) -> None: def test_apply_writes_distutils_cfg_for_detected_python_prefixes(tmp_path) -> None:
config = Configuration.model_validate(ONE_REPO_CONFIG) config = Configuration.model_validate(ONE_REPO_CONFIG)
src = tmp_path / "src" src = tmp_path / "src"
+104 -3
View File
@@ -6,7 +6,15 @@ CONFIG_WITH_REPO = {
"uuid": "12345678-1234-5678-1234-567812345678", "uuid": "12345678-1234-5678-1234-567812345678",
"organisation": "InterDiode", "organisation": "InterDiode",
"configurations": { "configurations": {
"rust": {"repositories": [{"name": "demo", "url": "https://example.invalid/"}]} "rust": {
"repositories": [
{
"name": "demo",
"url": "https://example.invalid/",
"is_sparse": True,
}
]
}
}, },
} }
@@ -16,8 +24,16 @@ CONFIG_WITH_TWO_REPOS = {
"configurations": { "configurations": {
"rust": { "rust": {
"repositories": [ "repositories": [
{"name": "demo", "url": "https://example.invalid/"}, {
{"name": "other", "url": "https://other.invalid/"}, "name": "demo",
"url": "https://example.invalid/",
"is_sparse": True,
},
{
"name": "other",
"url": "https://other.invalid/",
"is_sparse": True,
},
] ]
} }
}, },
@@ -61,6 +77,91 @@ def test_apply_writes_replacement_source(tmp_path) -> None:
) )
def test_apply_sanitizes_repository_names_with_special_characters(tmp_path) -> None:
data = {
"uuid": "12345678-1234-5678-1234-567812345678",
"organisation": "InterDiode",
"configurations": {
"rust": {
"repositories": [
{
"name": "weird.demo]\nname",
"url": "https://example.invalid/",
"is_sparse": True,
}
]
}
},
}
dst = tmp_path / "dst"
context = Context(config=make_config(data), dst=dst, src=None)
RustModule().apply(context)
config_toml = (dst / ".cargo/config.toml").read_text(encoding="utf-8")
assert config_toml == (
"[source.crates-io]\n"
'replace-with = "weird_demo__name"\n'
"\n"
"[source.weird_demo__name]\n"
'registry = "sparse+https://example.invalid/"\n'
)
def test_apply_does_not_prefix_url_when_not_sparse(tmp_path) -> None:
data = {
"uuid": "12345678-1234-5678-1234-567812345678",
"organisation": "InterDiode",
"configurations": {
"rust": {
"repositories": [{"name": "demo", "url": "https://example.invalid/"}]
}
},
}
dst = tmp_path / "dst"
context = Context(config=make_config(data), dst=dst, src=None)
RustModule().apply(context)
config_toml = (dst / ".cargo/config.toml").read_text(encoding="utf-8")
assert config_toml == (
"[source.crates-io]\n"
'replace-with = "demo"\n'
"\n"
"[source.demo]\n"
'registry = "https://example.invalid/"\n'
)
def test_apply_prefixes_url_with_sparse_when_configured(tmp_path) -> None:
dst = tmp_path / "dst"
context = Context(config=make_config(CONFIG_WITH_REPO), dst=dst, src=None)
RustModule().apply(context)
config_toml = (dst / ".cargo/config.toml").read_text(encoding="utf-8")
assert config_toml == (
"[source.crates-io]\n"
'replace-with = "demo"\n'
"\n"
"[source.demo]\n"
'registry = "sparse+https://example.invalid/"\n'
)
def test_apply_writes_cargo_config_for_each_configured_user(tmp_path) -> None:
dst = tmp_path / "dst"
context = Context(
config=make_config(CONFIG_WITH_REPO), dst=dst, src=None, users=["alice"]
)
RustModule().apply(context)
root_content = (dst / ".cargo/config.toml").read_text(encoding="utf-8")
alice_content = (dst / "home/alice/.cargo/config.toml").read_text(encoding="utf-8")
assert alice_content == root_content
def test_apply_writes_additional_repositories_as_named_registries(tmp_path) -> None: def test_apply_writes_additional_repositories_as_named_registries(tmp_path) -> None:
dst = tmp_path / "dst" dst = tmp_path / "dst"
context = Context(config=make_config(CONFIG_WITH_TWO_REPOS), dst=dst, src=None) context = Context(config=make_config(CONFIG_WITH_TWO_REPOS), dst=dst, src=None)
+110
View File
@@ -0,0 +1,110 @@
from id_configurator.users import PasswdEntry, parse_login_defs_uid_min, parse_passwd
PASSWD_CONTENT = (
"root:x:0:0:root:/root:/bin/bash\n"
"alice:x:1000:1000:Alice:/home/alice:/bin/bash\n"
"daemon:x:1:1:daemon:/usr/sbin:/usr/sbin/nologin\n"
"sync:x:4:65534:sync:/bin:/bin/sync\n"
"nobody:x:65534:65534:nobody:/nonexistent:/usr/sbin/false\n"
"# a comment\n"
"\n"
)
def test_parse_passwd_returns_one_entry_per_line() -> None:
entries = parse_passwd(PASSWD_CONTENT)
assert [entry.name for entry in entries] == [
"root",
"alice",
"daemon",
"sync",
"nobody",
]
def test_parse_passwd_skips_comments_and_blank_lines() -> None:
entries = parse_passwd(PASSWD_CONTENT)
assert len(entries) == 5
def test_parse_passwd_extracts_uid_home_and_shell() -> None:
entries = {entry.name: entry for entry in parse_passwd(PASSWD_CONTENT)}
assert entries["alice"].uid == 1000
assert entries["alice"].home == "/home/alice"
assert entries["alice"].shell == "/bin/bash"
def test_parse_passwd_skips_malformed_lines() -> None:
entries = parse_passwd(
"tooshort:x:0\nnotanumber:x:abc:0:x:/home/x:/bin/bash\n"
"root:x:0:0:root:/root:/bin/bash\n"
)
assert [entry.name for entry in entries] == ["root"]
def test_has_login_shell_true_for_regular_shell() -> None:
entry = PasswdEntry(name="alice", uid=1000, home="/home/alice", shell="/bin/bash") # noqa: B604
assert entry.has_login_shell is True
def test_has_login_shell_false_for_nologin_sync_or_false() -> None:
nologin = PasswdEntry(
name="daemon",
uid=1,
home="/usr/sbin",
shell="/usr/sbin/nologin", # noqa: B604
)
sync = PasswdEntry(name="sync", uid=4, home="/bin", shell="/bin/sync") # noqa: B604
false_shell = PasswdEntry(
name="nobody",
uid=65534,
home="/nonexistent",
shell="/usr/sbin/false", # noqa: B604
)
assert nologin.has_login_shell is False
assert sync.has_login_shell is False
assert false_shell.has_login_shell is False
def test_is_regular_user_true_for_human_account() -> None:
entry = PasswdEntry(name="alice", uid=1000, home="/home/alice", shell="/bin/bash") # noqa: B604
assert entry.is_regular_user() is True
def test_is_regular_user_true_for_root_despite_uid_zero() -> None:
entry = PasswdEntry(name="root", uid=0, home="/root", shell="/bin/bash") # noqa: B604
assert entry.is_regular_user() is True
def test_is_regular_user_false_for_root_with_disabled_shell() -> None:
entry = PasswdEntry(name="root", uid=0, home="/root", shell="/usr/sbin/nologin") # noqa: B604
assert entry.is_regular_user() is False
def test_is_regular_user_false_for_disabled_shell_despite_high_uid() -> None:
entry = PasswdEntry(
name="nobody",
uid=65534,
home="/nonexistent",
shell="/usr/sbin/false", # noqa: B604
)
assert entry.is_regular_user() is False
def test_is_regular_user_honors_custom_uid_min() -> None:
entry = PasswdEntry(name="svc", uid=500, home="/home/svc", shell="/bin/bash") # noqa: B604
assert entry.is_regular_user(uid_min=1000) is False
assert entry.is_regular_user(uid_min=100) is True
def test_parse_login_defs_uid_min_found() -> None:
content = "# comment\nUID_MIN\t500\nGID_MIN\t500\n"
assert parse_login_defs_uid_min(content) == 500
def test_parse_login_defs_uid_min_missing() -> None:
assert parse_login_defs_uid_min("GID_MIN\t500\n") is None
def test_parse_login_defs_uid_min_malformed_value() -> None:
assert parse_login_defs_uid_min("UID_MIN\tnot-a-number\n") is None
+31 -10
View File
@@ -313,10 +313,26 @@ def test_apply_matches_index_url_to_replace_with_unrelated_files_and_multiple_re
assert (dst / "etc/yum.repos.d/second.repo").exists() assert (dst / "etc/yum.repos.d/second.repo").exists()
def test_apply_sanitizes_repository_and_key_names_with_slash( def test_apply_sanitizes_key_name_with_slash(tmp_path, monkeypatch) -> None:
tmp_path, monkeypatch
) -> None:
mock_run = _mock_gpg_success(monkeypatch) mock_run = _mock_gpg_success(monkeypatch)
data = {
"uuid": "12345678-1234-5678-1234-567812345678",
"organisation": "InterDiode",
"configurations": {
"yum": {"keys": [{"name": "../../evil-key", "content": KEY_CONTENT}]}
},
}
dst = tmp_path / "dst"
context = Context(config=make_config(data), dst=dst, src=None)
YumModule().apply(context)
key_content = (dst / "etc/pki/rpm-gpg/RPM-GPG-KEY-.._.._evil-key").read_bytes()
assert key_content == b"fake-rpm-gpg-key-bytes"
mock_run.assert_called_once()
def test_apply_sanitizes_repository_name_with_disallowed_characters(tmp_path) -> None:
data = { data = {
"uuid": "12345678-1234-5678-1234-567812345678", "uuid": "12345678-1234-5678-1234-567812345678",
"organisation": "InterDiode", "organisation": "InterDiode",
@@ -324,11 +340,10 @@ def test_apply_sanitizes_repository_and_key_names_with_slash(
"yum": { "yum": {
"repositories": [ "repositories": [
{ {
"name": "../../evil-repo", "name": "weird/repo]here[bad",
"index-url": "https://example.invalid/yum", "index-url": "https://example.invalid/yum",
} }
], ]
"keys": [{"name": "../../evil-key", "content": KEY_CONTENT}],
} }
}, },
} }
@@ -337,10 +352,16 @@ def test_apply_sanitizes_repository_and_key_names_with_slash(
YumModule().apply(context) YumModule().apply(context)
assert (dst / "etc/yum.repos.d/.._.._evil-repo.repo").exists() repo_file = dst / "etc/yum.repos.d/weird_repo_here_bad.repo"
key_content = (dst / "etc/pki/rpm-gpg/RPM-GPG-KEY-.._.._evil-key").read_bytes() assert repo_file.exists()
assert key_content == b"fake-rpm-gpg-key-bytes" repo_content = repo_file.read_text(encoding="utf-8")
mock_run.assert_called_once() assert repo_content == (
"[weird_repo_here_bad]\n"
"name=weird_repo_here_bad\n"
"baseurl=https://example.invalid/yum\n"
"enabled=1\n"
"gpgcheck=0\n"
)
def test_import_key_raises_when_gpg_binary_missing(tmp_path, monkeypatch) -> None: def test_import_key_raises_when_gpg_binary_missing(tmp_path, monkeypatch) -> None: