diff --git a/id_configurator/cli.py b/id_configurator/cli.py index dc8f7f4..c52ad75 100644 --- a/id_configurator/cli.py +++ b/id_configurator/cli.py @@ -55,6 +55,17 @@ ConfigUrlCacertContentOption = Annotated[ ), ), ] +ConfigUrlInsecureOption = Annotated[ + bool, + typer.Option( + "--config-url-insecure", + help=( + "Disable TLS certificate verification when downloading --config-url " + "(INSECURE: only for bootstrapping from a server whose certificate " + "isn't trustable yet; overrides --config-url-cacert-content)." + ), + ), +] SrcOption = Annotated[ Path | None, typer.Option( @@ -89,13 +100,16 @@ def _load_configuration( config_env: str | None, config_url: str | None, config_url_cacert_content: str | None, + config_url_insecure: bool = False, ): if config is None and config_env is None and config_url is None: raise typer.BadParameter( "either --config, --config-env (or $ID_CONFIGURATOR_CONFIG) or " "--config-url must be provided" ) - return load_configuration(config, config_env, config_url, config_url_cacert_content) + return load_configuration( + config, config_env, config_url, config_url_cacert_content, config_url_insecure + ) @app.command() @@ -105,13 +119,14 @@ def apply( config_env: ConfigEnvOption = None, config_url: ConfigUrlOption = None, config_url_cacert_content: ConfigUrlCacertContentOption = None, + config_url_insecure: ConfigUrlInsecureOption = False, src: SrcOption = None, verbose: VerboseOption = False, ) -> None: """Apply a configuration to the destination directory.""" _configure_logging(verbose) configuration = _load_configuration( - config, config_env, config_url, config_url_cacert_content + config, config_env, config_url, config_url_cacert_content, config_url_insecure ) dst.mkdir(parents=True, exist_ok=True) context = Context(config=configuration, dst=dst, src=src) @@ -130,14 +145,15 @@ def validate( config_env: ConfigEnvOption = None, config_url: ConfigUrlOption = None, config_url_cacert_content: ConfigUrlCacertContentOption = None, + config_url_insecure: ConfigUrlInsecureOption = False, ) -> None: """Validate a configuration file without applying it.""" configuration = _load_configuration( - config, config_env, config_url, config_url_cacert_content - ) - typer.echo( - f"configuration {configuration.uuid} is valid (organisation: {configuration.organisation})" + config, config_env, config_url, config_url_cacert_content, config_url_insecure ) + uuid = configuration.uuid if configuration.uuid is not None else "" + organisation = configuration.organisation or "" + typer.echo(f"configuration {uuid} is valid (organisation: {organisation})") @app.command("list-modules") diff --git a/id_configurator/config.py b/id_configurator/config.py index aed9f19..d64db30 100644 --- a/id_configurator/config.py +++ b/id_configurator/config.py @@ -10,7 +10,7 @@ from pathlib import Path from uuid import UUID import yaml -from pydantic import BaseModel, ConfigDict, Field +from pydantic import BaseModel, ConfigDict, Field, field_validator _CONFIG_URL_TIMEOUT_SECONDS = 30 @@ -294,11 +294,28 @@ class ConfigurationsBlock(BaseModel): class Configuration(BaseModel): - uuid: UUID - organisation: str - description: str = "" + """`version` is the configuration format's own schema version (currently + "1.0"), independent of the id-configurator package version - it lets a + future, incompatible format change be told apart from this one. Omitting + it defaults to "1.0", the only version there currently is.""" + + version: str = "1.0" + uuid: UUID | None = None + organisation: str | None = None + description: str | None = None configurations: ConfigurationsBlock = Field(default_factory=ConfigurationsBlock) + @field_validator("version", mode="before") + @classmethod + def _coerce_version_to_str(cls, value: object) -> object: + """Accept a bare YAML number (`version: 1.0`) as well as a quoted + string (`version: "1.0"`): YAML parses the unquoted, natural-looking + form as a float, which would otherwise fail validation with a + confusing type error.""" + if isinstance(value, (int, float)) and not isinstance(value, bool): + return str(value) + return value + def parse_configuration(raw: str) -> Configuration: """Parse a configuration from a JSON or YAML string.""" @@ -309,17 +326,30 @@ def parse_configuration(raw: str) -> Configuration: return Configuration.model_validate(data) -def _fetch_config_url(url: str, cacert_content: str | None) -> str: +def _fetch_config_url( + url: str, cacert_content: str | None, insecure: bool = False +) -> 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). + + If `insecure` is set instead, certificate verification (and hostname + checking) is disabled entirely - e.g. to bootstrap from a server whose + certificate isn't trusted, or trustable, at all yet. `cacert_content` is + then ignored: there would be nothing left to verify it against. """ - context = ssl.create_default_context( - cadata=cacert_content if cacert_content is not None else None - ) + if insecure: + # Explicitly requested by the operator via --config-url-insecure. + context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + context.check_hostname = False + context.verify_mode = ssl.CERT_NONE + else: + 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: @@ -331,6 +361,7 @@ def load_configuration( config_env: str | None, config_url: str | None = None, config_url_cacert_content: str | None = None, + config_url_insecure: bool = False, ) -> Configuration: """Load a configuration from a file, from a base64-encoded string (typically an env var), or by downloading it from a URL.""" @@ -339,7 +370,9 @@ def load_configuration( 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) + raw = _fetch_config_url( + config_url, config_url_cacert_content, config_url_insecure + ) else: raise ValueError( "either --config, --config-env or --config-url must be provided" diff --git a/library/id_configurator.py b/library/id_configurator.py index d3c7d73..6c5ae1e 100644 --- a/library/id_configurator.py +++ b/library/id_configurator.py @@ -38,18 +38,19 @@ description: author: - InterDiode options: + version: + type: str + default: "1.0" + description: Configuration.version - the configuration format's own schema version. uuid: type: str - required: true - description: Configuration.uuid - a UUID identifying this configuration. + description: Configuration.uuid - a UUID identifying this configuration, optional. organisation: type: str - required: true - description: Configuration.organisation. + description: Configuration.organisation, optional. description: type: str - default: "" - description: Configuration.description. + description: Configuration.description, optional. certificates: type: list elements: dict @@ -467,9 +468,10 @@ def _name_url_repository_config() -> dict: ARGUMENT_SPEC = { - "uuid": {"type": "str", "required": True}, - "organisation": {"type": "str", "required": True}, - "description": {"type": "str", "default": ""}, + "version": {"type": "str", "default": "1.0"}, + "uuid": {"type": "str"}, + "organisation": {"type": "str"}, + "description": {"type": "str"}, "certificates": { "type": "list", "elements": "dict", @@ -596,6 +598,7 @@ except ImportError as exc: def _build_configuration_data(params: dict) -> dict: return { + "version": params["version"], "uuid": params["uuid"], "organisation": params["organisation"], "description": params["description"], diff --git a/tests/test_cli.py b/tests/test_cli.py index 8fbaf1d..6546fc3 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -30,6 +30,14 @@ def test_validate_valid_config(tmp_path) -> None: assert "is valid" in result.stdout +def test_validate_without_uuid_or_organisation(tmp_path) -> None: + config_path = write_config(tmp_path, {}) + result = runner.invoke(cli.app, ["validate", "--config", config_path]) + assert result.exit_code == 0 + assert "" in result.stdout + assert "" in result.stdout + + def test_validate_without_config_fails() -> None: result = runner.invoke(cli.app, ["validate"]) assert result.exit_code != 0 @@ -56,6 +64,24 @@ def test_validate_from_config_url(monkeypatch) -> None: assert "is valid" in result.stdout +def test_validate_from_config_url_insecure(monkeypatch) -> None: + mock_urlopen = _mock_urlopen(json.dumps(MINIMAL_CONFIG).encode("utf-8")) + monkeypatch.setattr("id_configurator.config.urllib.request.urlopen", mock_urlopen) + result = runner.invoke( + cli.app, + [ + "validate", + "--config-url", + "https://example.invalid/config.json", + "--config-url-insecure", + ], + ) + assert result.exit_code == 0 + assert "is valid" in result.stdout + _, kwargs = mock_urlopen.call_args + assert kwargs["context"].verify_mode.name == "CERT_NONE" + + def test_list_modules() -> None: result = runner.invoke(cli.app, ["list-modules"]) assert result.exit_code == 0 diff --git a/tests/test_config.py b/tests/test_config.py index e7a1027..ff88348 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -85,10 +85,38 @@ def test_parse_configuration_from_yaml() -> None: """ configuration = parse_configuration(raw) assert configuration.organisation == "InterDiode" - assert configuration.description == "" + 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") @@ -166,6 +194,44 @@ def test_load_configuration_from_url_with_custom_cacert(monkeypatch) -> None: 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"]