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
Files
26fcdd08e409e73480f22bc7de36d5b2dfb762a7
id_configurator/tests/test_java.py
T
d9pouces 35c3437343 feat: new configurations for APK/Java/Python/Yum 
Signed-off-by: d9pouces <github@19pouces.net>
2026-09-03 22:56:28 +02:00

390 lines
13 KiB
Python
Raw Blame History

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": NEW_CERT.public_bytes(Encoding.PEM),
}
]
},
}
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
Reference in New Issue View Git Blame Copy Permalink