InterDiode/id_configurator
1
0
Fork 0
mirror of https://github.com/d9pouces/id_configurator.git synced 2026-11-09 01:47:21 +01:00
Code Issues Packages Projects Releases Wiki Activity

feat: new configurations for APK/Java/Python/Yum

Browse Source 
Signed-off-by: d9pouces <github@19pouces.net>
This commit is contained in:
d9pouces
2026-09-02 06:58:20 +02:00
parent 48ec887808
commit 586f091eeb
22 changed files with 1804 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_apk.py
+104
View File
@@ -0,0 +1,104 @@
import base64
import hashlib
from id_configurator.config import Configuration
from id_configurator.context import Context
from id_configurator.modules.apk import ApkModule
KEY_BYTES = b"fake-apk-key-bytes"
KEY_FILENAME = f"{hashlib.sha256(KEY_BYTES).hexdigest()[:16]}.rsa.pub"
CONFIG_DATA = {
"uuid": "12345678-1234-5678-1234-567812345678",
"organisation": "InterDiode",
"configurations": {
"apk": {
"repositories": [
{
"url": "https://example.invalid/v3.19/main",
"key": base64.b64encode(KEY_BYTES).decode("ascii"),
},
{"url": "https://example.invalid/v3.19/community"},
]
}
},
}
def make_config() -> Configuration:
return Configuration.model_validate(CONFIG_DATA)
def test_is_relevant_false_without_src(tmp_path) -> None:
context = Context(config=make_config(), dst=tmp_path / "dst", src=None)
assert ApkModule().is_relevant(context) is False
def test_is_relevant_false_without_apk_dir(tmp_path) -> None:
src = tmp_path / "src"
src.mkdir()
context = Context(config=make_config(), dst=tmp_path / "dst", src=src)
assert ApkModule().is_relevant(context) is False
def test_is_relevant_true_with_apk_dir(tmp_path) -> None:
src = tmp_path / "src"
(src / "etc/apk").mkdir(parents=True)
context = Context(config=make_config(), dst=tmp_path / "dst", src=src)
assert ApkModule().is_relevant(context) is True
def test_apply_disables_existing_and_adds_configured_repositories(tmp_path) -> None:
src = tmp_path / "src"
apk_dir = src / "etc/apk"
apk_dir.mkdir(parents=True)
(apk_dir / "repositories").write_text(
"# comment\nhttps://old-mirror.invalid/v3.18/main\nhttps://old-mirror.invalid/v3.18/community\n",
encoding="utf-8",
)
dst = tmp_path / "dst"
context = Context(config=make_config(), dst=dst, src=src)
ApkModule().apply(context)
content = (dst / "etc/apk/repositories").read_text(encoding="utf-8")
assert content == (
"# comment\n"
"# https://old-mirror.invalid/v3.18/main\n"
"# https://old-mirror.invalid/v3.18/community\n"
"https://example.invalid/v3.19/main\n"
"https://example.invalid/v3.19/community\n"
)
key_content = (dst / "etc/apk/keys" / KEY_FILENAME).read_bytes()
assert key_content == KEY_BYTES
def test_apply_without_existing_repositories_file(tmp_path) -> None:
src = tmp_path / "src"
(src / "etc/apk").mkdir(parents=True)
dst = tmp_path / "dst"
context = Context(config=make_config(), dst=dst, src=src)
ApkModule().apply(context)
content = (dst / "etc/apk/repositories").read_text(encoding="utf-8")
assert (
content
== "https://example.invalid/v3.19/main\nhttps://example.invalid/v3.19/community\n"
)
def test_apply_writes_nothing_when_no_repositories_at_all(tmp_path) -> None:
empty_config = Configuration.model_validate(
{"uuid": "12345678-1234-5678-1234-567812345678", "organisation": "InterDiode"}
)
src = tmp_path / "src"
(src / "etc/apk").mkdir(parents=True)
dst = tmp_path / "dst"
context = Context(config=empty_config, dst=dst, src=src)
ApkModule().apply(context)
assert not (dst / "etc/apk/repositories").exists()
tests/test_apt.py
+13 -13
View File
@@ -34,21 +34,21 @@ def make_config() -> Configuration:
def test_is_relevant_false_without_src(tmp_path) -> None:
context = Context(config=make_config(), dst=tmp_path / "dst", src=None)
assert AptModule().is_relevant(context) is False # nosec B101
assert AptModule().is_relevant(context) is False
def test_is_relevant_false_without_apt_dir(tmp_path) -> None:
src = tmp_path / "src"
src.mkdir()
context = Context(config=make_config(), dst=tmp_path / "dst", src=src)
assert AptModule().is_relevant(context) is False # nosec B101
assert AptModule().is_relevant(context) is False
def test_is_relevant_true_with_apt_dir(tmp_path) -> None:
src = tmp_path / "src"
(src / "etc/apt").mkdir(parents=True)
context = Context(config=make_config(), dst=tmp_path / "dst", src=src)
assert AptModule().is_relevant(context) is True # nosec B101
assert AptModule().is_relevant(context) is True
def test_apply_disables_existing_sources_and_adds_configured_ones(tmp_path) -> None:
@@ -82,31 +82,31 @@ def test_apply_disables_existing_sources_and_adds_configured_ones(tmp_path) -> N
assert (
disabled_top_level
== "# already disabled\n# deb http://example.invalid/legacy stable main\n\n"
) # nosec B101
)
disabled_list = (dst / "etc/apt/sources.list.d/extra.list").read_text(
encoding="utf-8"
) # nosec B101
assert disabled_list == "# deb http://example.invalid/extra stable main\n" # nosec B101
)
assert disabled_list == "# deb http://example.invalid/extra stable main\n"
disabled_sources = (dst / "etc/apt/sources.list.d/docker.sources").read_text(
encoding="utf-8"
) # nosec B101
assert disabled_sources == ( # nosec B101
)
assert disabled_sources == (
"Types: deb\nURIs: http://example.invalid/docker\nSuites: stable\nEnabled: no"
"\n\n"
"Types: deb\nURIs: http://example.invalid/docker2\nSuites: stable\nEnabled: no\n"
)
assert not (dst / "etc/apt/sources.list.d/ignored.txt").exists() # nosec B101
assert not (dst / "etc/apt/sources.list.d/ignored.txt").exists()
keyring = (dst / "etc/apt/keyrings/interdiode.asc").read_bytes()
assert keyring == b"fake-key-bytes" # nosec B101
assert keyring == b"fake-key-bytes"
configured_source = (dst / "etc/apt/sources.list.d/interdiode.sources").read_text(
encoding="utf-8"
)
assert configured_source == ( # nosec B101
assert configured_source == (
"Types: deb\nURIs: http://example.invalid/debian\nSuites: stable\n"
"Signed-By: /etc/apt/keyrings/interdiode.asc\nEnabled: yes\n"
)
@@ -120,5 +120,5 @@ def test_apply_without_existing_top_level_sources_list(tmp_path) -> None:
AptModule().apply(context)
assert not (dst / "etc/apt/sources.list").exists() # nosec B101
assert (dst / "etc/apt/sources.list.d/interdiode.sources").exists() # nosec B101
assert not (dst / "etc/apt/sources.list").exists()
assert (dst / "etc/apt/sources.list.d/interdiode.sources").exists()
tests/test_certificates.py
+13 -13
View File
@@ -25,12 +25,12 @@ def make_config(with_certificate: bool) -> Configuration:
def test_is_relevant_true_when_certificates_configured(tmp_path) -> None:
context = Context(config=make_config(True), dst=tmp_path / "dst", src=None)
assert CertificatesModule().is_relevant(context) is True # nosec B101
assert CertificatesModule().is_relevant(context) is True
def test_is_relevant_false_when_no_certificates(tmp_path) -> None:
context = Context(config=make_config(False), dst=tmp_path / "dst", src=None)
assert CertificatesModule().is_relevant(context) is False # nosec B101
assert CertificatesModule().is_relevant(context) is False
def _mock_success(monkeypatch, stderr: str = "") -> MagicMock:
@@ -53,12 +53,12 @@ def test_apply_writes_certificate_and_calls_update_ca_certificates(
CertificatesModule().apply(context)
written = dst / "usr/local/share/ca-certificates/new-root-ca.crt"
assert written.read_bytes() == b"fake-cert-bytes" # nosec B101
assert written.read_bytes() == b"fake-cert-bytes"
mock_run.assert_called_once()
args = mock_run.call_args.args[0]
assert args[0] == "update-ca-certificates" # nosec B101
assert "--certsconf" not in args # nosec B101
assert "--certsdir" not in args # nosec B101
assert args[0] == "update-ca-certificates"
assert "--certsconf" not in args
assert "--certsdir" not in args
def test_apply_preserves_existing_local_certificates_from_src(
@@ -82,14 +82,14 @@ def test_apply_preserves_existing_local_certificates_from_src(
CertificatesModule().apply(context)
local_dst = dst / "usr/local/share/ca-certificates"
assert (local_dst / "existing.crt").read_bytes() == b"existing-bytes" # nosec B101
assert (local_dst / "new-root-ca.crt").read_bytes() == b"fake-cert-bytes" # nosec B101
assert (local_dst / "existing.crt").read_bytes() == b"existing-bytes"
assert (local_dst / "new-root-ca.crt").read_bytes() == b"fake-cert-bytes"
args = mock_run.call_args.args[0]
assert "--certsconf" in args # nosec B101
assert str(src / "etc/ca-certificates.conf") in args # nosec B101
assert "--certsdir" in args # nosec B101
assert str(src / "usr/share/ca-certificates") in args # nosec B101
assert "--certsconf" in args
assert str(src / "etc/ca-certificates.conf") in args
assert "--certsdir" in args
assert str(src / "usr/share/ca-certificates") in args
def test_update_ca_certificates_raises_when_binary_missing(
@@ -116,4 +116,4 @@ def test_update_ca_certificates_raises_on_failure(tmp_path, monkeypatch) -> None
context = Context(config=make_config(True), dst=tmp_path / "dst", src=None)
with pytest.raises(RuntimeError, match="update-ca-certificates failed"):
CertificatesModule().apply(context) # nosec B101
CertificatesModule().apply(context)
tests/test_cli.py
+22 -11
View File
@@ -32,20 +32,27 @@ def write_config(tmp_path, data) -> str:
def test_validate_valid_config(tmp_path) -> None:
config_path = write_config(tmp_path, MINIMAL_CONFIG)
result = runner.invoke(cli.app, ["validate", "--config", config_path])
assert result.exit_code == 0 # nosec B101
assert "is valid" in result.stdout # nosec B101
assert result.exit_code == 0
assert "is valid" in result.stdout
def test_validate_without_config_fails() -> None:
result = runner.invoke(cli.app, ["validate"])
assert result.exit_code != 0 # nosec B101
assert "either --config or --config-env" in result.output # nosec B101
assert result.exit_code != 0
assert "either --config or --config-env" in result.output
def test_list_modules() -> None:
result = runner.invoke(cli.app, ["list-modules"])
assert result.exit_code == 0 # nosec B101
assert set(result.stdout.splitlines()) == {"apt", "certificates"} # nosec B101
assert result.exit_code == 0
assert set(result.stdout.splitlines()) == {
"apk",
"apt",
"certificates",
"java",
"python",
"yum",
}
def test_apply_skips_irrelevant_modules(tmp_path, caplog) -> None:
@@ -55,9 +62,13 @@ def test_apply_skips_irrelevant_modules(tmp_path, caplog) -> None:
result = runner.invoke(
cli.app, ["apply", "--config", config_path, "--dst", str(dst), "-v"]
)
assert result.exit_code == 0 # nosec B101
assert "skipping module apt" in caplog.text # nosec B101
assert "skipping module certificates" in caplog.text # nosec B101
assert result.exit_code == 0
assert "skipping module apk" in caplog.text
assert "skipping module java" in caplog.text
assert "skipping module apt" in caplog.text
assert "skipping module certificates" in caplog.text
assert "skipping module yum" in caplog.text
assert "skipping module python" in caplog.text
def test_apply_runs_relevant_module(tmp_path, monkeypatch) -> None:
@@ -74,8 +85,8 @@ def test_apply_runs_relevant_module(tmp_path, monkeypatch) -> None:
cli.app, ["apply", "--config", config_path, "--dst", str(dst)]
)
assert result.exit_code == 0 # nosec B101
assert (dst / "usr/local/share/ca-certificates/my-ca.crt").read_bytes() == b"fake" # nosec B101
assert result.exit_code == 0
assert (dst / "usr/local/share/ca-certificates/my-ca.crt").read_bytes() == b"fake"
mock_run.assert_called_once()
tests/test_config.py
+18 -16
View File
@@ -36,7 +36,8 @@ FULL_CONFIG = {
"index-url": "http://example.invalid/yum",
"gpg-url": "http://example.invalid/gpg",
}
]
],
"keys": [{"name": "interdiode", "content": "ZmFrZQ=="}],
},
"python": {
"repositories": [
@@ -49,26 +50,27 @@ FULL_CONFIG = {
def test_parse_configuration_from_json() -> None:
configuration = parse_configuration(json.dumps(FULL_CONFIG))
assert isinstance(configuration, Configuration) # nosec B101
assert configuration.organisation == "InterDiode" # nosec B101
assert configuration.configurations.certificates[0].name == "my-ca" # nosec B101
assert isinstance(configuration, Configuration)
assert configuration.organisation == "InterDiode"
assert configuration.configurations.certificates[0].name == "my-ca"
assert (
configuration.configurations.yum.repositories[0].index_url
== "http://example.invalid/yum"
) # nosec B101
)
assert (
configuration.configurations.yum.repositories[0].gpg_url
== "http://example.invalid/gpg"
) # nosec B101
)
assert (
configuration.configurations.python.repositories[0].index_url
== "http://example.invalid/simple"
) # nosec B101
)
assert (
configuration.configurations.apt.sources[0].signed_by
== "/etc/apt/keyrings/interdiode.asc"
) # nosec B101
assert configuration.configurations.apt.keys[0].name == "interdiode" # nosec B101
)
assert configuration.configurations.apt.keys[0].name == "interdiode"
assert configuration.configurations.yum.keys[0].name == "interdiode"
def test_parse_configuration_from_yaml() -> None:
@@ -77,22 +79,22 @@ def test_parse_configuration_from_yaml() -> None:
organisation: InterDiode
"""
configuration = parse_configuration(raw)
assert configuration.organisation == "InterDiode" # nosec B101
assert configuration.description == "" # nosec B101
assert configuration.configurations.certificates == [] # nosec B101
assert configuration.organisation == "InterDiode"
assert configuration.description == ""
assert configuration.configurations.certificates == []
def test_load_configuration_from_file(tmp_path) -> None:
config_file = tmp_path / "config.json"
config_file.write_text(json.dumps(FULL_CONFIG), encoding="utf-8")
configuration = load_configuration(config_file, None)
assert configuration.organisation == "InterDiode" # nosec B101
assert configuration.organisation == "InterDiode"
def test_load_configuration_from_env() -> None:
encoded = base64.b64encode(json.dumps(FULL_CONFIG).encode("utf-8")).decode("ascii")
configuration = load_configuration(None, encoded)
assert configuration.organisation == "InterDiode" # nosec B101
assert configuration.organisation == "InterDiode"
def test_load_configuration_without_source_raises() -> None:
@@ -107,7 +109,7 @@ def test_apt_source_to_deb822_minimal() -> None:
assert (
source.to_deb822()
== "Types: deb\nURIs: http://example.invalid/debian\nSuites: stable\nEnabled: yes\n"
) # nosec B101
)
def test_apt_source_to_deb822_full() -> None:
@@ -138,7 +140,7 @@ def test_apt_source_to_deb822_full() -> None:
description="InterDiode APT repository",
)
rendered = source.to_deb822()
assert rendered.splitlines() == [ # nosec B101
assert rendered.splitlines() == [
"Types: deb deb-src",
"URIs: http://example.invalid/debian",
"Suites: stable",
tests/test_context.py
+25 -11
View File
@@ -21,7 +21,7 @@ def make_context(tmp_path, with_src: bool = True) -> Context:
def test_dst_path_within_destination(tmp_path) -> None:
context = make_context(tmp_path)
resolved = context.dst_path("etc/foo.conf")
assert resolved == context.dst / "etc" / "foo.conf" # nosec B101
assert resolved == context.dst / "etc" / "foo.conf"
def test_dst_path_rejects_escape(tmp_path) -> None:
@@ -32,46 +32,46 @@ def test_dst_path_rejects_escape(tmp_path) -> None:
def test_src_path_without_src(tmp_path) -> None:
context = make_context(tmp_path, with_src=False)
assert context.src_path("etc/foo.conf") is None # nosec B101
assert context.src_path("etc/foo.conf") is None
def test_src_path_with_src(tmp_path) -> None:
context = make_context(tmp_path)
resolved = context.src_path("etc/foo.conf")
assert resolved == context.src / "etc" / "foo.conf" # nosec B101
assert resolved == context.src / "etc" / "foo.conf"
def test_write_file_creates_parents_and_content(tmp_path) -> None:
context = make_context(tmp_path)
path = context.write_file("etc/nested/foo.conf", b"hello")
assert path.read_bytes() == b"hello" # nosec B101
assert path.stat().st_mode & 0o777 == 0o644 # nosec B101
assert path.read_bytes() == b"hello"
assert path.stat().st_mode & 0o777 == 0o644
def test_read_src_file_without_src(tmp_path) -> None:
context = make_context(tmp_path, with_src=False)
assert context.read_src_file("etc/foo.conf") is None # nosec B101
assert context.read_src_file("etc/foo.conf") is None
def test_read_src_file_missing(tmp_path) -> None:
context = make_context(tmp_path)
assert context.read_src_file("etc/missing.conf") is None # nosec B101
assert context.read_src_file("etc/missing.conf") is None
def test_read_src_file_present(tmp_path) -> None:
context = make_context(tmp_path)
(context.src / "foo.conf").write_bytes(b"existing")
assert context.read_src_file("foo.conf") == b"existing" # nosec B101
assert context.read_src_file("foo.conf") == b"existing"
def test_iter_src_dir_without_src(tmp_path) -> None:
context = make_context(tmp_path, with_src=False)
assert context.iter_src_dir("some/dir") == [] # nosec B101
assert context.iter_src_dir("some/dir") == []
def test_iter_src_dir_missing_dir(tmp_path) -> None:
context = make_context(tmp_path)
assert context.iter_src_dir("missing-dir") == [] # nosec B101
assert context.iter_src_dir("missing-dir") == []
def test_iter_src_dir_lists_only_files(tmp_path) -> None:
@@ -82,4 +82,18 @@ def test_iter_src_dir_lists_only_files(tmp_path) -> None:
(target / "b.crt").write_bytes(b"b")
(target / "subdir").mkdir()
result = context.iter_src_dir("certs")
assert [entry.name for entry in result] == ["a.crt", "b.crt"] # nosec B101
assert [entry.name for entry in result] == ["a.crt", "b.crt"]
def test_glob_src_without_src(tmp_path) -> None:
context = make_context(tmp_path, with_src=False)
assert context.glob_src("usr/lib/python3.*") == []
def test_glob_src_matches_pattern(tmp_path) -> None:
context = make_context(tmp_path)
(context.src / "usr/lib/python3.11").mkdir(parents=True)
(context.src / "usr/lib/python3.12").mkdir(parents=True)
(context.src / "usr/lib/python-legacy").mkdir(parents=True)
result = context.glob_src("usr/lib/python3.*")
assert [entry.name for entry in result] == ["python3.11", "python3.12"]
tests/test_java.py
+392
View File
@@ -0,0 +1,392 @@
import base64
import datetime
import jks
import pytest
from cryptography import x509
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives.serialization import (
BestAvailableEncryption,
Encoding,
pkcs12,
)
from cryptography.x509.oid import NameOID
from id_configurator.config import Configuration
from id_configurator.context import Context
from id_configurator.modules.java import JavaModule, _keystore_kind
def _self_signed_cert(common_name: str) -> x509.Certificate:
key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
subject = issuer = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, common_name)])
now = datetime.datetime.now(datetime.UTC)
return (
x509.CertificateBuilder()
.subject_name(subject)
.issuer_name(issuer)
.public_key(key.public_key())
.serial_number(x509.random_serial_number())
.not_valid_before(now)
.not_valid_after(now + datetime.timedelta(days=1))
.sign(key, hashes.SHA256())
)
NEW_CERT = _self_signed_cert("interdiode-root-ca")
EXISTING_CERT = _self_signed_cert("already-trusted-ca")
CONFIG_WITH_CERT = {
"uuid": "12345678-1234-5678-1234-567812345678",
"organisation": "InterDiode",
"configurations": {
"certificates": [
{
"name": "interdiode-root-ca",
"content": base64.b64encode(NEW_CERT.public_bytes(Encoding.PEM)).decode(
"ascii"
),
}
]
},
}
EMPTY_CONFIG = {
"uuid": "12345678-1234-5678-1234-567812345678",
"organisation": "InterDiode",
}
def make_config(data: dict) -> Configuration:
return Configuration.model_validate(data)
def _pkcs12_cacerts(
entries: list[tuple[str, x509.Certificate]] = (), password: bytes = b"changeit"
) -> bytes:
# serialize_java_truststore refuses an empty cert list, so seed with a placeholder when none given.
seeded = list(entries) or [("placeholder-ca", EXISTING_CERT)]
certs = [
pkcs12.PKCS12Certificate(cert, name.encode("utf-8")) for name, cert in seeded
]
return pkcs12.serialize_java_truststore(certs, BestAvailableEncryption(password))
def _jks_cacerts(entries: list[tuple[str, x509.Certificate]] = ()) -> bytes:
store = jks.KeyStore.new("jks", [])
for name, cert in entries:
store.entries[name] = jks.TrustedCertEntry.new(
name, cert.public_bytes(Encoding.DER)
)
return store.saves("changeit")
def test_is_relevant_false_without_certificates(tmp_path) -> None:
src = tmp_path / "src"
(src / "usr/lib/jvm/java-21/lib/security").mkdir(parents=True)
(src / "usr/lib/jvm/java-21/lib/security/cacerts").write_bytes(_pkcs12_cacerts())
context = Context(config=make_config(EMPTY_CONFIG), dst=tmp_path / "dst", src=src)
assert JavaModule().is_relevant(context) is False
def test_is_relevant_false_without_java_installation(tmp_path) -> None:
context = Context(
config=make_config(CONFIG_WITH_CERT), dst=tmp_path / "dst", src=None
)
assert JavaModule().is_relevant(context) is False
def test_is_relevant_true_with_certificate_and_cacerts(tmp_path) -> None:
src = tmp_path / "src"
(src / "usr/lib/jvm/java-21/lib/security").mkdir(parents=True)
(src / "usr/lib/jvm/java-21/lib/security/cacerts").write_bytes(_pkcs12_cacerts())
context = Context(
config=make_config(CONFIG_WITH_CERT), dst=tmp_path / "dst", src=src
)
assert JavaModule().is_relevant(context) is True
def test_apply_adds_certificate_to_pkcs12_and_jks_keystores(tmp_path) -> None:
src = tmp_path / "src"
pkcs12_dir = src / "usr/lib/jvm/java-21/lib/security"
jks_dir = src / "opt/java/openjdk8/jre/lib/security"
pkcs12_dir.mkdir(parents=True)
jks_dir.mkdir(parents=True)
(pkcs12_dir / "cacerts").write_bytes(
_pkcs12_cacerts([("already-trusted-ca", EXISTING_CERT)])
)
(jks_dir / "cacerts").write_bytes(
_jks_cacerts([("already-trusted-ca", EXISTING_CERT)])
)
dst = tmp_path / "dst"
context = Context(config=make_config(CONFIG_WITH_CERT), dst=dst, src=src)
JavaModule().apply(context)
pkcs12_result = pkcs12.load_pkcs12(
(dst / "usr/lib/jvm/java-21/lib/security/cacerts").read_bytes(), b"changeit"
)
pkcs12_aliases = {entry.friendly_name for entry in pkcs12_result.additional_certs}
assert pkcs12_aliases == {b"already-trusted-ca", b"interdiode-root-ca"}
jks_result = jks.KeyStore.loads(
(dst / "opt/java/openjdk8/jre/lib/security/cacerts").read_bytes(), "changeit"
)
assert set(jks_result.certs.keys()) == {"already-trusted-ca", "interdiode-root-ca"}
assert jks_result.certs["interdiode-root-ca"].cert == NEW_CERT.public_bytes(
Encoding.DER
)
def test_apply_skips_keystore_with_wrong_password(tmp_path, caplog) -> None:
src = tmp_path / "src"
good_dir = src / "usr/lib/jvm/java-21/lib/security"
bad_dir = src / "opt/java/other/lib/security"
good_dir.mkdir(parents=True)
bad_dir.mkdir(parents=True)
(good_dir / "cacerts").write_bytes(_pkcs12_cacerts())
(bad_dir / "cacerts").write_bytes(_pkcs12_cacerts(password=b"not-changeit"))
dst = tmp_path / "dst"
context = Context(config=make_config(CONFIG_WITH_CERT), dst=dst, src=src)
with caplog.at_level("WARNING", logger="id_configurator"):
JavaModule().apply(context)
assert (dst / "usr/lib/jvm/java-21/lib/security/cacerts").exists()
assert not (dst / "opt/java/other/lib/security/cacerts").exists()
assert "skipping java keystore" in caplog.text
@pytest.mark.parametrize(
("content", "expected"),
[
(jks.MAGIC_NUMBER_JKS + b"rest", "jks"),
(jks.MAGIC_NUMBER_JCEKS + b"rest", "jks"),
(b"\x30\x82\x00\x00rest", "pkcs12"),
],
)
def test_keystore_kind_detection(content, expected) -> None:
assert _keystore_kind(content) == expected
ONE_REPO_CONFIG = {
"uuid": "12345678-1234-5678-1234-567812345678",
"organisation": "InterDiode",
"configurations": {
"java": {
"repositories": [
{"name": "interdiode", "url": "https://example.invalid/maven"}
]
}
},
}
TWO_REPOS_CONFIG = {
"uuid": "12345678-1234-5678-1234-567812345678",
"organisation": "InterDiode",
"configurations": {
"java": {
"repositories": [
{"name": "interdiode", "url": "https://example.invalid/maven"},
{
"name": "interdiode-extra",
"url": "https://example2.invalid/maven",
},
]
}
},
}
def test_is_relevant_true_with_repositories_and_no_src(tmp_path) -> None:
context = Context(
config=make_config(ONE_REPO_CONFIG), dst=tmp_path / "dst", src=None
)
assert JavaModule().is_relevant(context) is True
def test_apply_without_certificates_only_writes_repository_files(tmp_path) -> None:
dst = tmp_path / "dst"
context = Context(config=make_config(ONE_REPO_CONFIG), dst=dst, src=None)
JavaModule().apply(context)
assert not (dst / "usr").exists()
assert (dst / "etc/maven/settings.xml").exists()
def test_apply_writes_maven_settings(tmp_path) -> None:
dst = tmp_path / "dst"
context = Context(config=make_config(TWO_REPOS_CONFIG), dst=dst, src=None)
JavaModule().apply(context)
settings = (dst / "etc/maven/settings.xml").read_text(encoding="utf-8")
assert settings == (
'<?xml version="1.0" encoding="UTF-8"?>\n'
'<settings xmlns="http://maven.apache.org/SETTINGS/1.1.0" '
'xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" '
'xsi:schemaLocation="http://maven.apache.org/SETTINGS/1.1.0 '
'http://maven.apache.org/xsd/settings-1.1.0.xsd">\n'
" <profiles>\n"
" <profile>\n"
" <id>id-configurator</id>\n"
" <activation>\n"
" <activeByDefault>true</activeByDefault>\n"
" </activation>\n"
" <repositories>\n"
" <repository>\n"
" <id>interdiode</id>\n"
" <url>https://example.invalid/maven</url>\n"
" </repository>\n"
" <repository>\n"
" <id>interdiode-extra</id>\n"
" <url>https://example2.invalid/maven</url>\n"
" </repository>\n"
" </repositories>\n"
" <pluginRepositories>\n"
" <pluginRepository>\n"
" <id>interdiode</id>\n"
" <url>https://example.invalid/maven</url>\n"
" </pluginRepository>\n"
" <pluginRepository>\n"
" <id>interdiode-extra</id>\n"
" <url>https://example2.invalid/maven</url>\n"
" </pluginRepository>\n"
" </pluginRepositories>\n"
" </profile>\n"
" </profiles>\n"
" <activeProfiles>\n"
" <activeProfile>id-configurator</activeProfile>\n"
" </activeProfiles>\n"
" <mirrors>\n"
" <mirror>\n"
" <id>interdiode</id>\n"
" <url>https://example.invalid/maven</url>\n"
" <mirrorOf>*</mirrorOf>\n"
" </mirror>\n"
" </mirrors>\n"
"</settings>\n"
)
def test_apply_writes_gradle_init_script(tmp_path) -> None:
dst = tmp_path / "dst"
context = Context(config=make_config(TWO_REPOS_CONFIG), dst=dst, src=None)
JavaModule().apply(context)
script = (
dst / "root/.gradle/init.d/id-configurator-repositories.gradle"
).read_text(encoding="utf-8")
assert script == (
"allprojects {\n"
" repositories {\n"
' maven { url = uri("https://example.invalid/maven") }\n'
' maven { url = uri("https://example2.invalid/maven") }\n'
" }\n"
"}\n"
"\n"
"settingsEvaluated { settings ->\n"
" settings.pluginManagement {\n"
" repositories {\n"
' maven { url = uri("https://example.invalid/maven") }\n'
' maven { url = uri("https://example2.invalid/maven") }\n'
" }\n"
" }\n"
"}\n"
)
def test_apply_writes_ivy_settings(tmp_path) -> None:
dst = tmp_path / "dst"
context = Context(config=make_config(TWO_REPOS_CONFIG), dst=dst, src=None)
JavaModule().apply(context)
ivysettings = (dst / "root/.ivy2/ivysettings.xml").read_text(encoding="utf-8")
assert ivysettings == (
'<?xml version="1.0" encoding="UTF-8"?>\n'
"<ivysettings>\n"
' <settings defaultResolver="id-configurator" />\n'
" <resolvers>\n"
' <chain name="id-configurator">\n'
' <ibiblio name="interdiode" m2compatible="true" '
'root="https://example.invalid/maven" />\n'
' <ibiblio name="interdiode-extra" m2compatible="true" '
'root="https://example2.invalid/maven" />\n'
" </chain>\n"
" </resolvers>\n"
"</ivysettings>\n"
)
def test_apply_writes_sbt_repositories(tmp_path) -> None:
dst = tmp_path / "dst"
context = Context(config=make_config(TWO_REPOS_CONFIG), dst=dst, src=None)
JavaModule().apply(context)
repositories = (dst / "root/.sbt/repositories").read_text(encoding="utf-8")
assert repositories == (
"[repositories]\n"
" local\n"
" interdiode: https://example.invalid/maven\n"
" interdiode-extra: https://example2.invalid/maven\n"
)
def test_apply_writes_lein_profiles(tmp_path) -> None:
dst = tmp_path / "dst"
context = Context(config=make_config(TWO_REPOS_CONFIG), dst=dst, src=None)
JavaModule().apply(context)
profiles = (dst / "root/.lein/profiles.clj").read_text(encoding="utf-8")
assert profiles == (
"{:user {:repositories "
'[["interdiode" {:url "https://example.invalid/maven"}] '
'["interdiode-extra" {:url "https://example2.invalid/maven"}]]}}\n'
)
def test_apply_with_certificates_and_repositories_writes_both(tmp_path) -> None:
src = tmp_path / "src"
(src / "usr/lib/jvm/java-21/lib/security").mkdir(parents=True)
(src / "usr/lib/jvm/java-21/lib/security/cacerts").write_bytes(_pkcs12_cacerts())
combined_config = {
**CONFIG_WITH_CERT,
"configurations": {
**CONFIG_WITH_CERT["configurations"],
"java": ONE_REPO_CONFIG["configurations"]["java"],
},
}
dst = tmp_path / "dst"
context = Context(config=make_config(combined_config), dst=dst, src=src)
JavaModule().apply(context)
assert (dst / "usr/lib/jvm/java-21/lib/security/cacerts").exists()
assert (dst / "etc/maven/settings.xml").exists()
def test_apply_writes_maven_settings_to_detected_installations(tmp_path) -> None:
src = tmp_path / "src"
(src / "usr/share/maven/bin").mkdir(parents=True)
(src / "usr/share/maven/bin/mvn").write_text("#!/bin/sh\n")
(src / "opt/apache-maven-3.9.6/bin").mkdir(parents=True)
(src / "opt/apache-maven-3.9.6/bin/mvn").write_text("#!/bin/sh\n")
dst = tmp_path / "dst"
context = Context(config=make_config(ONE_REPO_CONFIG), dst=dst, src=src)
JavaModule().apply(context)
global_settings = (dst / "etc/maven/settings.xml").read_bytes()
assert (dst / "usr/share/maven/conf/settings.xml").read_bytes() == global_settings
assert (
dst / "opt/apache-maven-3.9.6/conf/settings.xml"
).read_bytes() == global_settings
tests/test_modules_base.py
+1 -1
View File
@@ -16,4 +16,4 @@ def test_register_and_all_modules_are_isolated(monkeypatch) -> None:
register(DummyModule)
assert all_modules() == [DummyModule] # nosec B101
assert all_modules() == [DummyModule]
tests/test_package.py
+2 -2
View File
@@ -6,8 +6,8 @@ from unittest.mock import MagicMock
def test_package_exposes_app_and_main() -> None:
import id_configurator
assert id_configurator.main is id_configurator.cli.main # nosec B101
assert id_configurator.app is id_configurator.cli.app # nosec B101
assert id_configurator.main is id_configurator.cli.main
assert id_configurator.app is id_configurator.cli.app
def test_dunder_main_module_does_not_launch_app_when_imported(monkeypatch) -> None:
tests/test_python.py
+139
View File
@@ -0,0 +1,139 @@
from id_configurator.config import Configuration
from id_configurator.context import Context
from id_configurator.modules.python import PythonModule
ONE_REPO_CONFIG = {
"uuid": "12345678-1234-5678-1234-567812345678",
"organisation": "InterDiode",
"configurations": {
"python": {
"repositories": [
{"name": "interdiode", "index-url": "https://example.invalid/simple"}
]
}
},
}
TWO_REPOS_CONFIG = {
"uuid": "12345678-1234-5678-1234-567812345678",
"organisation": "InterDiode",
"configurations": {
"python": {
"repositories": [
{"name": "interdiode", "index-url": "https://example.invalid/simple"},
{
"name": "interdiode-extra",
"index-url": "https://example2.invalid/simple",
},
]
}
},
}
def test_is_relevant_false_without_repositories(tmp_path) -> None:
config = Configuration.model_validate(
{"uuid": "12345678-1234-5678-1234-567812345678", "organisation": "InterDiode"}
)
context = Context(config=config, dst=tmp_path / "dst", src=None)
assert PythonModule().is_relevant(context) is False
def test_is_relevant_true_with_repositories(tmp_path) -> None:
config = Configuration.model_validate(ONE_REPO_CONFIG)
context = Context(config=config, dst=tmp_path / "dst", src=None)
assert PythonModule().is_relevant(context) is True
def test_apply_single_repository(tmp_path) -> None:
config = Configuration.model_validate(ONE_REPO_CONFIG)
dst = tmp_path / "dst"
context = Context(config=config, dst=dst, src=None)
PythonModule().apply(context)
pip_conf = (dst / "etc/pip.conf").read_text(encoding="utf-8")
assert pip_conf == (
"[global]\nindex-url = https://example.invalid/simple\ncert = /etc/ssl/certs/ca-certificates.crt\n"
)
uv_toml = (dst / "etc/uv/uv.toml").read_text(encoding="utf-8")
assert uv_toml == (
'native-tls = true\n\n[[index]]\nname = "interdiode"\nurl = "https://example.invalid/simple"\ndefault = true\n'
)
poetry_config = (dst / "root/.config/pypoetry/config.toml").read_text(
encoding="utf-8"
)
assert poetry_config == (
"[repositories.interdiode]\n"
'url = "https://example.invalid/simple"\n'
"\n"
"[certificates.interdiode]\n"
'cert = "/etc/ssl/certs/ca-certificates.crt"\n'
)
assert not (dst / "usr/lib").exists()
def test_apply_multiple_repositories(tmp_path) -> None:
config = Configuration.model_validate(TWO_REPOS_CONFIG)
dst = tmp_path / "dst"
context = Context(config=config, dst=dst, src=None)
PythonModule().apply(context)
pip_conf = (dst / "etc/pip.conf").read_text(encoding="utf-8")
assert pip_conf == (
"[global]\n"
"index-url = https://example.invalid/simple\n"
"extra-index-url =\n"
" https://example2.invalid/simple\n"
"cert = /etc/ssl/certs/ca-certificates.crt\n"
)
uv_toml = (dst / "etc/uv/uv.toml").read_text(encoding="utf-8")
assert uv_toml == (
"native-tls = true\n\n"
"[[index]]\n"
'name = "interdiode"\n'
'url = "https://example.invalid/simple"\n'
"default = true\n\n"
"[[index]]\n"
'name = "interdiode-extra"\n'
'url = "https://example2.invalid/simple"\n'
)
def test_apply_writes_distutils_cfg_for_detected_python_prefixes(tmp_path) -> None:
config = Configuration.model_validate(ONE_REPO_CONFIG)
src = tmp_path / "src"
(src / "usr/lib/python3.11").mkdir(parents=True)
(src / "usr/local/lib/python3.12").mkdir(parents=True)
(src / "usr/lib/not-python").mkdir(parents=True)
dst = tmp_path / "dst"
context = Context(config=config, dst=dst, src=src)
PythonModule().apply(context)
content_311 = (dst / "usr/lib/python3.11/distutils/distutils.cfg").read_text(
encoding="utf-8"
)
assert content_311 == "[easy_install]\nindex_url = https://example.invalid/simple\n"
content_312 = (dst / "usr/local/lib/python3.12/distutils/distutils.cfg").read_text(
encoding="utf-8"
)
assert content_312 == "[easy_install]\nindex_url = https://example.invalid/simple\n"
assert not (dst / "usr/lib/not-python").exists()
def test_apply_without_src_skips_distutils_cfg(tmp_path) -> None:
config = Configuration.model_validate(ONE_REPO_CONFIG)
dst = tmp_path / "dst"
context = Context(config=config, dst=dst, src=None)
PythonModule().apply(context)
assert not (dst / "usr").exists()
tests/test_yum.py
+194
View File
@@ -0,0 +1,194 @@
import base64
import subprocess
from configparser import ConfigParser
from unittest.mock import MagicMock
import pytest
from id_configurator.config import Configuration
from id_configurator.context import Context
from id_configurator.modules.yum import YumModule
KEY_CONTENT = base64.b64encode(b"fake-rpm-gpg-key-bytes").decode("ascii")
CONFIG_WITH_REPO_AND_KEY = {
"uuid": "12345678-1234-5678-1234-567812345678",
"organisation": "InterDiode",
"configurations": {
"yum": {
"repositories": [
{
"name": "interdiode",
"index-url": "https://example.invalid/yum",
"gpg-url": "https://example.invalid/RPM-GPG-KEY-interdiode",
}
],
"keys": [{"name": "interdiode", "content": KEY_CONTENT}],
}
},
}
CONFIG_WITH_UNSIGNED_REPO = {
"uuid": "12345678-1234-5678-1234-567812345678",
"organisation": "InterDiode",
"configurations": {
"yum": {
"repositories": [
{"name": "unsigned", "index-url": "https://example.invalid/yum"}
]
}
},
}
EMPTY_CONFIG = {
"uuid": "12345678-1234-5678-1234-567812345678",
"organisation": "InterDiode",
}
def make_config(data: dict) -> Configuration:
return Configuration.model_validate(data)
def test_is_relevant_false_when_empty(tmp_path) -> None:
context = Context(config=make_config(EMPTY_CONFIG), dst=tmp_path / "dst", src=None)
assert YumModule().is_relevant(context) is False
def test_is_relevant_true_with_repository(tmp_path) -> None:
context = Context(
config=make_config(CONFIG_WITH_UNSIGNED_REPO), dst=tmp_path / "dst", src=None
)
assert YumModule().is_relevant(context) is True
def test_is_relevant_true_with_existing_repo_file(tmp_path) -> None:
src = tmp_path / "src"
repos_dir = src / "etc/yum.repos.d"
repos_dir.mkdir(parents=True)
(repos_dir / "old.repo").write_text(
"[old]\nname=old\nbaseurl=https://old.invalid\nenabled=1\n", encoding="utf-8"
)
context = Context(config=make_config(EMPTY_CONFIG), dst=tmp_path / "dst", src=src)
assert YumModule().is_relevant(context) is True
def test_is_relevant_false_with_only_unrelated_src_files(tmp_path) -> None:
src = tmp_path / "src"
repos_dir = src / "etc/yum.repos.d"
repos_dir.mkdir(parents=True)
(repos_dir / "README").write_text("not a repo file\n", encoding="utf-8")
context = Context(config=make_config(EMPTY_CONFIG), dst=tmp_path / "dst", src=src)
assert YumModule().is_relevant(context) is False
def _mock_gpg_success(monkeypatch) -> MagicMock:
mock_run = MagicMock(
return_value=subprocess.CompletedProcess(
args=[], returncode=0, stdout=b"", stderr=b""
)
)
monkeypatch.setattr("id_configurator.modules.yum.subprocess.run", mock_run)
return mock_run
def test_apply_writes_signed_repo_and_imports_key(tmp_path, monkeypatch) -> None:
mock_run = _mock_gpg_success(monkeypatch)
dst = tmp_path / "dst"
context = Context(config=make_config(CONFIG_WITH_REPO_AND_KEY), dst=dst, src=None)
YumModule().apply(context)
repo_content = (dst / "etc/yum.repos.d/interdiode.repo").read_text(encoding="utf-8")
assert repo_content == (
"[interdiode]\n"
"name=interdiode\n"
"baseurl=https://example.invalid/yum\n"
"enabled=1\n"
"gpgcheck=1\n"
"gpgkey=https://example.invalid/RPM-GPG-KEY-interdiode\n"
)
key_content = (dst / "etc/pki/rpm-gpg/RPM-GPG-KEY-interdiode").read_bytes()
assert key_content == b"fake-rpm-gpg-key-bytes"
mock_run.assert_called_once()
assert mock_run.call_args.kwargs["input"] == b"fake-rpm-gpg-key-bytes"
def test_apply_unsigned_repo_disables_gpgcheck(tmp_path) -> None:
dst = tmp_path / "dst"
context = Context(config=make_config(CONFIG_WITH_UNSIGNED_REPO), dst=dst, src=None)
YumModule().apply(context)
repo_content = (dst / "etc/yum.repos.d/unsigned.repo").read_text(encoding="utf-8")
assert repo_content == (
"[unsigned]\nname=unsigned\nbaseurl=https://example.invalid/yum\nenabled=1\ngpgcheck=0\n"
)
def test_apply_disables_existing_repos_before_adding_configured_ones(tmp_path) -> None:
src = tmp_path / "src"
repos_dir = src / "etc/yum.repos.d"
repos_dir.mkdir(parents=True)
(repos_dir / "multi.repo").write_text(
"[old-one]\nname=old-one\nbaseurl=https://old-one.invalid\nenabled=1\ngpgcheck=1\n"
"\n"
"[old-two]\nname=old-two\nbaseurl=https://old-two.invalid\nenabled=1\n",
encoding="utf-8",
)
(repos_dir / "ignored.txt").write_text("not a repo file\n", encoding="utf-8")
# A same-named existing file must end up enabled again once the configured repo overwrites it.
(repos_dir / "unsigned.repo").write_text(
"[unsigned]\nname=unsigned\nbaseurl=https://stale.invalid\nenabled=1\n",
encoding="utf-8",
)
dst = tmp_path / "dst"
context = Context(config=make_config(CONFIG_WITH_UNSIGNED_REPO), dst=dst, src=src)
YumModule().apply(context)
disabled_multi = (dst / "etc/yum.repos.d/multi.repo").read_text(encoding="utf-8")
assert "[old-one]" in disabled_multi
assert "[old-two]" in disabled_multi
parser = ConfigParser(interpolation=None)
parser.read_string(disabled_multi)
assert parser["old-one"]["enabled"] == "0"
assert parser["old-two"]["enabled"] == "0"
assert not (dst / "etc/yum.repos.d/ignored.txt").exists()
overridden = (dst / "etc/yum.repos.d/unsigned.repo").read_text(encoding="utf-8")
assert (
overridden
== "[unsigned]\nname=unsigned\nbaseurl=https://example.invalid/yum\nenabled=1\ngpgcheck=0\n"
)
def test_import_key_raises_when_gpg_binary_missing(tmp_path, monkeypatch) -> None:
monkeypatch.setattr(
"id_configurator.modules.yum.subprocess.run",
MagicMock(side_effect=FileNotFoundError()),
)
context = Context(
config=make_config(CONFIG_WITH_REPO_AND_KEY), dst=tmp_path / "dst", src=None
)
with pytest.raises(RuntimeError, match="gpg is not available"):
YumModule().apply(context)
def test_import_key_raises_on_invalid_key(tmp_path, monkeypatch) -> None:
mock_run = MagicMock(
return_value=subprocess.CompletedProcess(
args=[], returncode=2, stdout=b"", stderr=b"gpg: no valid data found"
)
)
monkeypatch.setattr("id_configurator.modules.yum.subprocess.run", mock_run)
context = Context(
config=make_config(CONFIG_WITH_REPO_AND_KEY), dst=tmp_path / "dst", src=None
)
with pytest.raises(RuntimeError, match="invalid GPG key interdiode"):
YumModule().apply(context)