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 == "" 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" 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_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