Files
id_configurator/tests/test_config.py
T

331 lines
11 KiB
Python

import base64
import datetime
import json
from unittest.mock import MagicMock
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 Encoding
from cryptography.x509.oid import NameOID
from id_configurator.config import (
AptSource,
Configuration,
load_configuration,
parse_configuration,
)
FULL_CONFIG = {
"uuid": "12345678-1234-5678-1234-567812345678",
"organisation": "InterDiode",
"description": "a test configuration",
"configurations": {
"certificates": [{"name": "my-ca", "content": "ZmFrZQ=="}],
"apt": {
"sources": [
{
"name": "interdiode",
"types": ["deb"],
"uris": ["http://example.invalid/debian"],
"suites": ["stable"],
"components": ["main", "contrib"],
"signed-by": "/etc/apt/keyrings/interdiode.asc",
}
],
},
"yum": {
"repositories": [
{
"name": "interdiode",
"index-url": "http://example.invalid/yum",
"gpg-url": "http://example.invalid/gpg",
}
],
"keys": [{"name": "interdiode", "content": "ZmFrZQ=="}],
},
"python": {
"repositories": [
{"name": "interdiode", "index-url": "http://example.invalid/simple"}
]
},
},
}
def test_parse_configuration_from_json() -> None:
configuration = parse_configuration(json.dumps(FULL_CONFIG))
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"
)
assert (
configuration.configurations.yum.repositories[0].gpg_url
== "http://example.invalid/gpg"
)
assert (
configuration.configurations.python.repositories[0].index_url
== "http://example.invalid/simple"
)
assert (
configuration.configurations.apt.sources[0].signed_by
== "/etc/apt/keyrings/interdiode.asc"
)
assert configuration.configurations.yum.keys[0].name == "interdiode"
def test_parse_configuration_from_yaml() -> None:
raw = """
uuid: 12345678-1234-5678-1234-567812345678
organisation: InterDiode
"""
configuration = parse_configuration(raw)
assert configuration.organisation == "InterDiode"
assert configuration.description is None
assert configuration.configurations.certificates == []
def test_parse_configuration_without_uuid_or_organisation() -> None:
configuration = parse_configuration(json.dumps({}))
assert configuration.uuid is None
assert configuration.organisation is None
assert configuration.version == "1.0"
assert configuration.description is None
def test_parse_configuration_defaults_version_to_1_0() -> None:
configuration = parse_configuration(json.dumps(FULL_CONFIG))
assert configuration.version == "1.0"
def test_parse_configuration_with_explicit_version() -> None:
configuration = parse_configuration(json.dumps({**FULL_CONFIG, "version": "1.0"}))
assert configuration.version == "1.0"
def test_parse_configuration_coerces_unquoted_yaml_version_to_str() -> None:
raw = """
version: 1.0
uuid: 12345678-1234-5678-1234-567812345678
organisation: InterDiode
"""
configuration = parse_configuration(raw)
assert configuration.version == "1.0"
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"
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"
def test_load_configuration_without_source_raises() -> None:
with pytest.raises(
ValueError, match="either --config, --config-env or --config-url"
):
load_configuration(None, None)
def _mock_urlopen(content: bytes) -> MagicMock:
response = MagicMock()
response.read.return_value = content
response.__enter__.return_value = response
response.__exit__.return_value = False
return MagicMock(return_value=response)
def _self_signed_ca_pem() -> str:
key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
subject = issuer = x509.Name(
[x509.NameAttribute(NameOID.COMMON_NAME, "test-config-url-ca")]
)
now = datetime.datetime.now(datetime.UTC)
cert = (
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())
)
return cert.public_bytes(Encoding.PEM).decode("ascii")
def test_load_configuration_from_url(monkeypatch) -> None:
mock_urlopen = _mock_urlopen(json.dumps(FULL_CONFIG).encode("utf-8"))
monkeypatch.setattr("id_configurator.config.urllib.request.urlopen", mock_urlopen)
configuration = load_configuration(
None, None, "https://example.invalid/config.json"
)
assert configuration.organisation == "InterDiode"
args, kwargs = mock_urlopen.call_args
assert args[0] == "https://example.invalid/config.json"
assert kwargs["timeout"] == 30
assert kwargs["context"].verify_mode.name == "CERT_REQUIRED"
def test_load_configuration_from_url_with_custom_cacert(monkeypatch) -> None:
mock_urlopen = _mock_urlopen(json.dumps(FULL_CONFIG).encode("utf-8"))
monkeypatch.setattr("id_configurator.config.urllib.request.urlopen", mock_urlopen)
cacert_content = _self_signed_ca_pem()
configuration = load_configuration(
None, None, "https://example.invalid/config.json", cacert_content
)
assert configuration.organisation == "InterDiode"
_, kwargs = mock_urlopen.call_args
stats = kwargs["context"].cert_store_stats()
assert stats["x509"] == 1
def test_load_configuration_from_url_insecure_disables_verification(
monkeypatch,
) -> None:
mock_urlopen = _mock_urlopen(json.dumps(FULL_CONFIG).encode("utf-8"))
monkeypatch.setattr("id_configurator.config.urllib.request.urlopen", mock_urlopen)
configuration = load_configuration(
None,
None,
"https://example.invalid/config.json",
None,
True,
)
assert configuration.organisation == "InterDiode"
_, kwargs = mock_urlopen.call_args
assert kwargs["context"].verify_mode.name == "CERT_NONE"
assert kwargs["context"].check_hostname is False
def test_load_configuration_from_url_insecure_ignores_cacert(monkeypatch) -> None:
mock_urlopen = _mock_urlopen(json.dumps(FULL_CONFIG).encode("utf-8"))
monkeypatch.setattr("id_configurator.config.urllib.request.urlopen", mock_urlopen)
cacert_content = _self_signed_ca_pem()
configuration = load_configuration(
None,
None,
"https://example.invalid/config.json",
cacert_content,
True,
)
assert configuration.organisation == "InterDiode"
_, kwargs = mock_urlopen.call_args
assert kwargs["context"].verify_mode.name == "CERT_NONE"
def test_apt_source_to_deb822_minimal() -> None:
source = AptSource(
name="interdiode", uris=["http://example.invalid/debian"], suites=["stable"]
)
assert (
source.to_deb822()
== "Types: deb\nURIs: http://example.invalid/debian\nSuites: stable\nEnabled: yes\n"
)
def test_apt_source_to_deb822_full() -> None:
source = AptSource(
name="interdiode",
types=["deb", "deb-src"],
uris=["http://example.invalid/debian"],
suites=["stable"],
components=["main", "contrib"],
architectures=["amd64"],
languages=["en"],
targets=["deb"],
enabled=False,
trusted=True,
**{
"by-hash": True,
"allow-insecure": False,
"allow-weak": False,
"allow-downgrade-to-insecure": False,
"check-valid-until": True,
"valid-until-min": "10",
"valid-until-max": "1w",
"check-date": True,
"date-max-future": "1d",
},
pdiffs=True,
description="InterDiode APT repository",
)
rendered = source.to_deb822(signed_by_path="/etc/apt/keyrings/interdiode.asc")
assert rendered.splitlines() == [
"Types: deb deb-src",
"URIs: http://example.invalid/debian",
"Suites: stable",
"Components: main contrib",
"Architectures: amd64",
"Languages: en",
"Targets: deb",
"Signed-By: /etc/apt/keyrings/interdiode.asc",
"Trusted: yes",
"PDiffs: yes",
"By-Hash: yes",
"Allow-Insecure: no",
"Allow-Weak: no",
"Allow-Downgrade-To-Insecure: no",
"Check-Valid-Until: yes",
"Valid-Until-Min: 10",
"Valid-Until-Max: 1w",
"Check-Date: yes",
"Date-Max-Future: 1d",
"Description: InterDiode APT repository",
"Enabled: no",
]
def test_apt_source_to_deb822_folds_inline_signed_by_key() -> None:
source = AptSource(
name="interdiode", uris=["http://example.invalid/debian"], suites=["stable"]
)
armored_key = (
"-----BEGIN PGP PUBLIC KEY BLOCK-----\n"
"\n"
"mQENBGqh08IBCAC0\n"
"-----END PGP PUBLIC KEY BLOCK-----\n"
)
rendered = source.to_deb822(signed_by_armored_key=armored_key)
assert rendered == (
"Types: deb\n"
"URIs: http://example.invalid/debian\n"
"Suites: stable\n"
"Signed-By:\n"
" -----BEGIN PGP PUBLIC KEY BLOCK-----\n"
" .\n"
" mQENBGqh08IBCAC0\n"
" -----END PGP PUBLIC KEY BLOCK-----\n"
"Enabled: yes\n"
)
def test_apt_source_to_deb822_signed_by_armored_key_takes_precedence() -> None:
source = AptSource(
name="interdiode", uris=["http://example.invalid/debian"], suites=["stable"]
)
rendered = source.to_deb822(
signed_by_armored_key="-----BEGIN PGP PUBLIC KEY BLOCK-----\nabc\n-----END PGP PUBLIC KEY BLOCK-----\n",
signed_by_path="/etc/apt/keyrings/unused.gpg",
)
assert "Signed-By:\n" in rendered
assert "unused.gpg" not in rendered