From 88380b6ab6f0555b5ae3448f83b310120b906eb3 Mon Sep 17 00:00:00 2001 From: d9pouces Date: Sun, 6 Sep 2026 23:08:58 +0200 Subject: [PATCH] refactor: remove unused NPM_TO_COPY variable from build.py Signed-off-by: d9pouces --- id_configurator/cli.py | 46 ++++++++++++++++++++++---- id_configurator/config.py | 35 ++++++++++++++++++-- tests/test_cli.py | 22 ++++++++++++- tests/test_config.py | 68 ++++++++++++++++++++++++++++++++++++++- 4 files changed, 159 insertions(+), 12 deletions(-) diff --git a/id_configurator/cli.py b/id_configurator/cli.py index 36097e0..dc8f7f4 100644 --- a/id_configurator/cli.py +++ b/id_configurator/cli.py @@ -38,6 +38,23 @@ ConfigEnvOption = Annotated[ help="Base64-encoded JSON/YAML configuration.", ), ] +ConfigUrlOption = Annotated[ + str | None, + typer.Option( + "--config-url", + help="URL to download a JSON or YAML configuration from.", + ), +] +ConfigUrlCacertContentOption = Annotated[ + str | None, + typer.Option( + "--config-url-cacert-content", + help=( + "PEM content of the CA certificate to trust when downloading " + "--config-url over HTTPS (defaults to the system's trust store)." + ), + ), +] SrcOption = Annotated[ Path | None, typer.Option( @@ -67,12 +84,18 @@ def _configure_logging(verbose: bool) -> None: ) -def _load_configuration(config: Path | None, config_env: str | None): - if config is None and config_env is None: +def _load_configuration( + config: Path | None, + config_env: str | None, + config_url: str | None, + config_url_cacert_content: str | None, +): + if config is None and config_env is None and config_url is None: raise typer.BadParameter( - "either --config or --config-env (or $ID_CONFIGURATOR_CONFIG) must be provided" + "either --config, --config-env (or $ID_CONFIGURATOR_CONFIG) or " + "--config-url must be provided" ) - return load_configuration(config, config_env) + return load_configuration(config, config_env, config_url, config_url_cacert_content) @app.command() @@ -80,12 +103,16 @@ def apply( dst: DstOption, config: ConfigFileOption = None, config_env: ConfigEnvOption = None, + config_url: ConfigUrlOption = None, + config_url_cacert_content: ConfigUrlCacertContentOption = None, src: SrcOption = None, verbose: VerboseOption = False, ) -> None: """Apply a configuration to the destination directory.""" _configure_logging(verbose) - configuration = _load_configuration(config, config_env) + configuration = _load_configuration( + config, config_env, config_url, config_url_cacert_content + ) dst.mkdir(parents=True, exist_ok=True) context = Context(config=configuration, dst=dst, src=src) for module_cls in all_modules(): @@ -99,10 +126,15 @@ def apply( @app.command() def validate( - config: ConfigFileOption = None, config_env: ConfigEnvOption = None + config: ConfigFileOption = None, + config_env: ConfigEnvOption = None, + config_url: ConfigUrlOption = None, + config_url_cacert_content: ConfigUrlCacertContentOption = None, ) -> None: """Validate a configuration file without applying it.""" - configuration = _load_configuration(config, config_env) + configuration = _load_configuration( + config, config_env, config_url, config_url_cacert_content + ) typer.echo( f"configuration {configuration.uuid} is valid (organisation: {configuration.organisation})" ) diff --git a/id_configurator/config.py b/id_configurator/config.py index 3681240..2c2c153 100644 --- a/id_configurator/config.py +++ b/id_configurator/config.py @@ -4,12 +4,16 @@ from __future__ import annotations import base64 import json +import ssl +import urllib.request from pathlib import Path from uuid import UUID import yaml from pydantic import BaseModel, ConfigDict, Field +_CONFIG_URL_TIMEOUT_SECONDS = 30 + class NamedContent(BaseModel): """A named piece of content, generally base64-encoded (a certificate, a GPG key, ...).""" @@ -253,14 +257,39 @@ def parse_configuration(raw: str) -> Configuration: return Configuration.model_validate(data) +def _fetch_config_url(url: str, cacert_content: str | None) -> str: + """Download a JSON or YAML configuration from `url`. + + If `cacert_content` is given, it - and only it, not the system's default + trust store - is used to verify an HTTPS connection: for an internal + endpoint signed by a private CA the running system doesn't otherwise + trust yet (e.g. before the `certificates` module has ever run). + """ + context = ssl.create_default_context( + cadata=cacert_content if cacert_content is not None else None + ) + with urllib.request.urlopen( # nosec B310 (url is an operator-provided CLI argument) + url, timeout=_CONFIG_URL_TIMEOUT_SECONDS, context=context + ) as response: + return response.read().decode("utf-8") + + def load_configuration( - config_file: Path | None, config_env: str | None + config_file: Path | None, + config_env: str | None, + config_url: str | None = None, + config_url_cacert_content: str | None = None, ) -> Configuration: - """Load a configuration from a file, or from a base64-encoded string (typically an env var).""" + """Load a configuration from a file, from a base64-encoded string + (typically an env var), or by downloading it from a URL.""" if config_file is not None: raw = config_file.read_text(encoding="utf-8") elif config_env is not None: raw = base64.b64decode(config_env).decode("utf-8") + elif config_url is not None: + raw = _fetch_config_url(config_url, config_url_cacert_content) else: - raise ValueError("either --config or --config-env must be provided") + raise ValueError( + "either --config, --config-env or --config-url must be provided" + ) return parse_configuration(raw) diff --git a/tests/test_cli.py b/tests/test_cli.py index 954d663..8fbaf1d 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -33,7 +33,27 @@ def test_validate_valid_config(tmp_path) -> None: def test_validate_without_config_fails() -> None: result = runner.invoke(cli.app, ["validate"]) assert result.exit_code != 0 - assert "either --config or --config-env" in result.output + assert "either --config, --config-env" in result.output + + +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 test_validate_from_config_url(monkeypatch) -> None: + monkeypatch.setattr( + "id_configurator.config.urllib.request.urlopen", + _mock_urlopen(json.dumps(MINIMAL_CONFIG).encode("utf-8")), + ) + result = runner.invoke( + cli.app, ["validate", "--config-url", "https://example.invalid/config.json"] + ) + assert result.exit_code == 0 + assert "is valid" in result.stdout def test_list_modules() -> None: diff --git a/tests/test_config.py b/tests/test_config.py index 4efdb90..d30b748 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1,7 +1,14 @@ 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, @@ -98,10 +105,69 @@ def test_load_configuration_from_env() -> None: def test_load_configuration_without_source_raises() -> None: - with pytest.raises(ValueError, match="either --config or --config-env"): + 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"]