mirror of
https://github.com/d9pouces/id_configurator.git
synced 2026-11-09 01:47:21 +01:00
Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
204d648477
|
|||
|
eff9ca66d2
|
|||
|
776cdc3b17
|
+22
-6
@@ -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 "<no uuid>"
|
||||
organisation = configuration.organisation or "<no organisation>"
|
||||
typer.echo(f"configuration {uuid} is valid (organisation: {organisation})")
|
||||
|
||||
|
||||
@app.command("list-modules")
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -3,6 +3,7 @@ import logging
|
||||
import typer
|
||||
|
||||
from id_tools.build import app as build_app
|
||||
from id_tools.publication import app as publication_app
|
||||
from id_tools.test import app as test_app
|
||||
|
||||
SUCCESS = 25
|
||||
@@ -44,5 +45,6 @@ def main():
|
||||
|
||||
app.add_typer(build_app, name="build")
|
||||
app.add_typer(test_app, name="test")
|
||||
app.add_typer(publication_app, name="publication")
|
||||
if __name__ == "__main__":
|
||||
app()
|
||||
|
||||
@@ -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"],
|
||||
|
||||
+1
-1
@@ -45,7 +45,7 @@ maintainers = [
|
||||
name = "id-configurator"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.13"
|
||||
version = "2026.9.7"
|
||||
version = "2026.9.20"
|
||||
|
||||
[project.scripts]
|
||||
id-configurator = "id_configurator:main"
|
||||
|
||||
@@ -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 "<no uuid>" in result.stdout
|
||||
assert "<no organisation>" 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
|
||||
|
||||
+67
-1
@@ -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"]
|
||||
|
||||
Reference in New Issue
Block a user