mirror of
https://github.com/d9pouces/id_configurator.git
synced 2026-11-09 01:47:21 +01:00
d9pouces
417 lines
14 KiB
Python
417 lines
14 KiB
Python
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)
|
|
|
|
# The written truststore's trust-only entries are unencrypted (matching
|
|
# the JDK's own cacerts), so they load with no password at all - the
|
|
# password a caller would use for a source-of-truth JDK-style store.
|
|
pkcs12_result = pkcs12.load_pkcs12(
|
|
(dst / "usr/lib/jvm/java-21/lib/security/cacerts").read_bytes(), None
|
|
)
|
|
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_marks_non_https_lein_repository_as_insecure(tmp_path) -> None:
|
|
data = {
|
|
"uuid": "12345678-1234-5678-1234-567812345678",
|
|
"organisation": "InterDiode",
|
|
"configurations": {
|
|
"java": {
|
|
"repositories": [
|
|
{"name": "interdiode", "url": "http://example.invalid/maven"}
|
|
]
|
|
}
|
|
},
|
|
}
|
|
dst = tmp_path / "dst"
|
|
context = Context(config=make_config(data), dst=dst, src=None)
|
|
|
|
JavaModule().apply(context)
|
|
|
|
profiles = (dst / "root/.lein/profiles.clj").read_text(encoding="utf-8")
|
|
assert profiles == (
|
|
"{:user {:repositories "
|
|
'[["interdiode" {:url "http://example.invalid/maven" :insecure true}]]}}\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
|