refactor: remove unused NPM_TO_COPY variable from build.py

Signed-off-by: d9pouces <github@19pouces.net>

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})"
     )

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)

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:

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"]