Compare commits

...

7 Commits

Author SHA1 Message Date
copilot-swe-agent[bot]
8487e8ee1c Initial plan 2025-11-27 09:49:37 +00:00
Stefan Agner
a351b09757 ruff format 2025-11-27 10:48:47 +01:00
Stefan Agner
743b1c3b92 Apply suggestions from code review
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
2025-11-27 10:38:24 +01:00
Stefan Agner
5f6d7c230f Refactor registry credential extraction into shared helper
Extract duplicate logic for determining which registry matches an image
into a shared `get_registry_for_image()` method in `DockerConfig`. This
method is now used by both `DockerInterface._get_credentials()` and
`AddonBuild.get_docker_config_json()`.

Move `DOCKER_HUB` and `IMAGE_WITH_HOST` constants to `docker/const.py`
to avoid circular imports between manager.py and interface.py.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
2025-11-27 10:27:59 +01:00
Stefan Agner
e7be2bfd0d Apply suggestions from code review
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
2025-11-27 10:14:20 +01:00
Mike Degatano
8486354eba Fix pylint errors 2025-11-26 14:05:30 -05:00
Stefan Agner
65b10c1931 Pass registry credentials to add-on build for private base images
When building add-ons that use a base image from a private registry,
the build would fail because credentials configured via the Supervisor
API were not passed to the Docker-in-Docker build container.

This fix:
- Adds get_docker_config_json() to generate a Docker config.json with
  registry credentials for the base image
- Creates a temporary config file and mounts it into the build container
  at /root/.docker/config.json so BuildKit can authenticate when pulling
  the base image
- Cleans up the temporary file after build completes

Fixes #6354

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
2025-11-26 17:43:50 +01:00
7 changed files with 283 additions and 34 deletions

View File

@@ -2,7 +2,9 @@
from __future__ import annotations
import base64
from functools import cached_property
import json
from pathlib import Path
from typing import TYPE_CHECKING, Any
@@ -12,12 +14,15 @@ from ..const import (
ATTR_ARGS,
ATTR_BUILD_FROM,
ATTR_LABELS,
ATTR_PASSWORD,
ATTR_SQUASH,
ATTR_USERNAME,
FILE_SUFFIX_CONFIGURATION,
META_ADDON,
SOCKET_DOCKER,
)
from ..coresys import CoreSys, CoreSysAttributes
from ..docker.const import DOCKER_HUB
from ..docker.interface import MAP_ARCH
from ..exceptions import ConfigurationFileError, HassioArchNotFound
from ..utils.common import FileConfiguration, find_one_filetype
@@ -122,8 +127,39 @@ class AddonBuild(FileConfiguration, CoreSysAttributes):
except HassioArchNotFound:
return False
def get_docker_config_json(self) -> str | None:
"""Generate Docker config.json content with registry credentials for base image.
Returns a JSON string with registry credentials for the base image's registry,
or None if no matching registry is configured.
"""
# Early return before accessing base_image to avoid unnecessary arch lookup
if not self.sys_docker.config.registries:
return None
registry = self.sys_docker.config.get_registry_for_image(self.base_image)
if not registry:
return None
stored = self.sys_docker.config.registries[registry]
username = stored[ATTR_USERNAME]
password = stored[ATTR_PASSWORD]
# Docker config.json uses base64-encoded "username:password" for auth
auth_string = base64.b64encode(f"{username}:{password}".encode()).decode()
# Use the actual registry URL for the key
# Docker Hub uses "https://index.docker.io/v1/" as the key
registry_key = (
"https://index.docker.io/v1/" if registry == DOCKER_HUB else registry
)
config = {"auths": {registry_key: {"auth": auth_string}}}
return json.dumps(config)
def get_docker_args(
self, version: AwesomeVersion, image_tag: str
self, version: AwesomeVersion, image_tag: str, docker_config_path: Path | None
) -> dict[str, Any]:
"""Create a dict with Docker run args."""
dockerfile_path = self.get_dockerfile().relative_to(self.addon.path_location)
@@ -172,12 +208,24 @@ class AddonBuild(FileConfiguration, CoreSysAttributes):
self.addon.path_location
)
volumes = {
SOCKET_DOCKER: {"bind": "/var/run/docker.sock", "mode": "rw"},
addon_extern_path: {"bind": "/addon", "mode": "ro"},
}
# Mount Docker config with registry credentials if available
if docker_config_path:
docker_config_extern_path = self.sys_config.local_to_extern_path(
docker_config_path
)
volumes[docker_config_extern_path] = {
"bind": "/root/.docker/config.json",
"mode": "ro",
}
return {
"command": build_cmd,
"volumes": {
SOCKET_DOCKER: {"bind": "/var/run/docker.sock", "mode": "rw"},
addon_extern_path: {"bind": "/addon", "mode": "ro"},
},
"volumes": volumes,
"working_dir": "/addon",
}

View File

@@ -7,6 +7,7 @@ from ipaddress import IPv4Address
import logging
import os
from pathlib import Path
import tempfile
from typing import TYPE_CHECKING, cast
import aiodocker
@@ -705,12 +706,38 @@ class DockerAddon(DockerInterface):
with suppress(docker.errors.NotFound):
self.sys_docker.containers.get(builder_name).remove(force=True, v=True)
result = self.sys_docker.run_command(
ADDON_BUILDER_IMAGE,
version=builder_version_tag,
name=builder_name,
**build_env.get_docker_args(version, addon_image_tag),
)
# Generate Docker config with registry credentials for base image if needed
docker_config_path: Path | None = None
docker_config_content = build_env.get_docker_config_json()
temp_dir: tempfile.TemporaryDirectory | None = None
try:
if docker_config_content:
# Create temporary directory for docker config
temp_dir = tempfile.TemporaryDirectory(
prefix="hassio_build_", dir=self.sys_config.path_tmp
)
docker_config_path = Path(temp_dir.name) / "config.json"
docker_config_path.write_text(
docker_config_content, encoding="utf-8"
)
_LOGGER.debug(
"Created temporary Docker config for build at %s",
docker_config_path,
)
result = self.sys_docker.run_command(
ADDON_BUILDER_IMAGE,
version=builder_version_tag,
name=builder_name,
**build_env.get_docker_args(
version, addon_image_tag, docker_config_path
),
)
finally:
# Clean up temporary directory
if temp_dir:
temp_dir.cleanup()
logs = result.output.decode("utf-8")

View File

@@ -15,6 +15,12 @@ from ..const import MACHINE_ID
RE_RETRYING_DOWNLOAD_STATUS = re.compile(r"Retrying in \d+ seconds?")
# Docker Hub registry identifier
DOCKER_HUB = "hub.docker.com"
# Regex to match images with a registry host (e.g., ghcr.io/org/image)
IMAGE_WITH_HOST = re.compile(r"^((?:[a-z0-9]+(?:-[a-z0-9]+)*\.)+[a-z]{2,})\/.+")
class Capabilities(StrEnum):
"""Linux Capabilities."""

View File

@@ -8,7 +8,6 @@ from collections.abc import Awaitable
from contextlib import suppress
from http import HTTPStatus
import logging
import re
from time import time
from typing import Any, cast
from uuid import uuid4
@@ -46,16 +45,13 @@ from ..jobs.decorator import Job
from ..jobs.job_group import JobGroup
from ..resolution.const import ContextType, IssueType, SuggestionType
from ..utils.sentry import async_capture_exception
from .const import ContainerState, PullImageLayerStage, RestartPolicy
from .const import DOCKER_HUB, ContainerState, PullImageLayerStage, RestartPolicy
from .manager import CommandReturn, PullLogEntry
from .monitor import DockerContainerStateEvent
from .stats import DockerStats
_LOGGER: logging.Logger = logging.getLogger(__name__)
IMAGE_WITH_HOST = re.compile(r"^((?:[a-z0-9]+(?:-[a-z0-9]+)*\.)+[a-z]{2,})\/.+")
DOCKER_HUB = "hub.docker.com"
MAP_ARCH: dict[CpuArch | str, str] = {
CpuArch.ARMV7: "linux/arm/v7",
CpuArch.ARMHF: "linux/arm/v6",
@@ -180,25 +176,16 @@ class DockerInterface(JobGroup, ABC):
return self.meta_config.get("Healthcheck")
def _get_credentials(self, image: str) -> dict:
"""Return a dictionay with credentials for docker login."""
registry = None
"""Return a dictionary with credentials for docker login."""
credentials = {}
matcher = IMAGE_WITH_HOST.match(image)
# Custom registry
if matcher:
if matcher.group(1) in self.sys_docker.config.registries:
registry = matcher.group(1)
credentials[ATTR_REGISTRY] = registry
# If no match assume "dockerhub" as registry
elif DOCKER_HUB in self.sys_docker.config.registries:
registry = DOCKER_HUB
registry = self.sys_docker.config.get_registry_for_image(image)
if registry:
stored = self.sys_docker.config.registries[registry]
credentials[ATTR_USERNAME] = stored[ATTR_USERNAME]
credentials[ATTR_PASSWORD] = stored[ATTR_PASSWORD]
if registry != DOCKER_HUB:
credentials[ATTR_REGISTRY] = registry
_LOGGER.debug(
"Logging in to %s as %s",

View File

@@ -49,7 +49,7 @@ from ..exceptions import (
)
from ..utils.common import FileConfiguration
from ..validate import SCHEMA_DOCKER_CONFIG
from .const import LABEL_MANAGED
from .const import DOCKER_HUB, IMAGE_WITH_HOST, LABEL_MANAGED
from .monitor import DockerMonitor
from .network import DockerNetwork
@@ -202,6 +202,27 @@ class DockerConfig(FileConfiguration):
"""Return credentials for docker registries."""
return self._data.get(ATTR_REGISTRIES, {})
def get_registry_for_image(self, image: str) -> str | None:
"""Return the registry name if credentials are available for the image.
Matches the image against configured registries and returns the registry
name if found, or None if no matching credentials are configured.
"""
if not self.registries:
return None
# Check if image uses a custom registry (e.g., ghcr.io/org/image)
matcher = IMAGE_WITH_HOST.match(image)
if matcher:
registry = matcher.group(1)
if registry in self.registries:
return registry
# If no registry prefix, check for Docker Hub credentials
elif DOCKER_HUB in self.registries:
return DOCKER_HUB
return None
class DockerAPI(CoreSysAttributes):
"""Docker Supervisor wrapper.

View File

@@ -1,5 +1,8 @@
"""Test addon build."""
import base64
import json
from pathlib import Path
from unittest.mock import PropertyMock, patch
from awesomeversion import AwesomeVersion
@@ -7,6 +10,7 @@ from awesomeversion import AwesomeVersion
from supervisor.addons.addon import Addon
from supervisor.addons.build import AddonBuild
from supervisor.coresys import CoreSys
from supervisor.docker.const import DOCKER_HUB
from tests.common import is_in_list
@@ -29,7 +33,7 @@ async def test_platform_set(coresys: CoreSys, install_addon_ssh: Addon):
),
):
args = await coresys.run_in_executor(
build.get_docker_args, AwesomeVersion("latest"), "test-image:latest"
build.get_docker_args, AwesomeVersion("latest"), "test-image:latest", None
)
assert is_in_list(["--platform", "linux/amd64"], args["command"])
@@ -53,7 +57,7 @@ async def test_dockerfile_evaluation(coresys: CoreSys, install_addon_ssh: Addon)
),
):
args = await coresys.run_in_executor(
build.get_docker_args, AwesomeVersion("latest"), "test-image:latest"
build.get_docker_args, AwesomeVersion("latest"), "test-image:latest", None
)
assert is_in_list(["--file", "Dockerfile"], args["command"])
@@ -81,7 +85,7 @@ async def test_dockerfile_evaluation_arch(coresys: CoreSys, install_addon_ssh: A
),
):
args = await coresys.run_in_executor(
build.get_docker_args, AwesomeVersion("latest"), "test-image:latest"
build.get_docker_args, AwesomeVersion("latest"), "test-image:latest", None
)
assert is_in_list(["--file", "Dockerfile.aarch64"], args["command"])
@@ -117,3 +121,158 @@ async def test_build_invalid(coresys: CoreSys, install_addon_ssh: Addon):
),
):
assert not await build.is_valid()
async def test_docker_config_no_registries(coresys: CoreSys, install_addon_ssh: Addon):
"""Test docker config generation when no registries configured."""
build = await AddonBuild(coresys, install_addon_ssh).load_config()
# No registries configured by default
assert build.get_docker_config_json() is None
async def test_docker_config_no_matching_registry(
coresys: CoreSys, install_addon_ssh: Addon
):
"""Test docker config generation when registry doesn't match base image."""
build = await AddonBuild(coresys, install_addon_ssh).load_config()
# Configure a registry that doesn't match the base image
# pylint: disable-next=protected-access
coresys.docker.config._data["registries"] = {
"some.other.registry": {"username": "user", "password": "pass"}
}
with (
patch.object(
type(coresys.arch), "supported", new=PropertyMock(return_value=["amd64"])
),
patch.object(
type(coresys.arch), "default", new=PropertyMock(return_value="amd64")
),
):
# Base image is ghcr.io/home-assistant/... which doesn't match
assert build.get_docker_config_json() is None
async def test_docker_config_matching_registry(
coresys: CoreSys, install_addon_ssh: Addon
):
"""Test docker config generation when registry matches base image."""
build = await AddonBuild(coresys, install_addon_ssh).load_config()
# Configure ghcr.io registry which matches the default base image
# pylint: disable-next=protected-access
coresys.docker.config._data["registries"] = {
"ghcr.io": {"username": "testuser", "password": "testpass"}
}
with (
patch.object(
type(coresys.arch), "supported", new=PropertyMock(return_value=["amd64"])
),
patch.object(
type(coresys.arch), "default", new=PropertyMock(return_value="amd64")
),
):
config_json = build.get_docker_config_json()
assert config_json is not None
config = json.loads(config_json)
assert "auths" in config
assert "ghcr.io" in config["auths"]
# Verify base64-encoded credentials
expected_auth = base64.b64encode(b"testuser:testpass").decode()
assert config["auths"]["ghcr.io"]["auth"] == expected_auth
async def test_docker_config_docker_hub(coresys: CoreSys, install_addon_ssh: Addon):
"""Test docker config generation for Docker Hub registry."""
build = await AddonBuild(coresys, install_addon_ssh).load_config()
# Configure Docker Hub registry
# pylint: disable-next=protected-access
coresys.docker.config._data["registries"] = {
DOCKER_HUB: {"username": "hubuser", "password": "hubpass"}
}
# Mock base_image to return a Docker Hub image (no registry prefix)
with patch.object(
type(build),
"base_image",
new=PropertyMock(return_value="library/alpine:latest"),
):
config_json = build.get_docker_config_json()
assert config_json is not None
config = json.loads(config_json)
# Docker Hub uses special URL as key
assert "https://index.docker.io/v1/" in config["auths"]
expected_auth = base64.b64encode(b"hubuser:hubpass").decode()
assert config["auths"]["https://index.docker.io/v1/"]["auth"] == expected_auth
async def test_docker_args_with_config_path(coresys: CoreSys, install_addon_ssh: Addon):
"""Test docker args include config volume when path provided."""
build = await AddonBuild(coresys, install_addon_ssh).load_config()
with (
patch.object(
type(coresys.arch), "supported", new=PropertyMock(return_value=["amd64"])
),
patch.object(
type(coresys.arch), "default", new=PropertyMock(return_value="amd64")
),
patch.object(
type(coresys.config),
"local_to_extern_path",
side_effect=lambda p: f"/extern{p}",
),
):
config_path = Path("/data/supervisor/tmp/config.json")
args = await coresys.run_in_executor(
build.get_docker_args,
AwesomeVersion("latest"),
"test-image:latest",
config_path,
)
# Check that config is mounted
assert "/extern/data/supervisor/tmp/config.json" in args["volumes"]
assert (
args["volumes"]["/extern/data/supervisor/tmp/config.json"]["bind"]
== "/root/.docker/config.json"
)
assert args["volumes"]["/extern/data/supervisor/tmp/config.json"]["mode"] == "ro"
async def test_docker_args_without_config_path(
coresys: CoreSys, install_addon_ssh: Addon
):
"""Test docker args don't include config volume when no path provided."""
build = await AddonBuild(coresys, install_addon_ssh).load_config()
with (
patch.object(
type(coresys.arch), "supported", new=PropertyMock(return_value=["amd64"])
),
patch.object(
type(coresys.arch), "default", new=PropertyMock(return_value="amd64")
),
patch.object(
type(coresys.config),
"local_to_extern_path",
return_value="/addon/path/on/host",
),
):
args = await coresys.run_in_executor(
build.get_docker_args, AwesomeVersion("latest"), "test-image:latest", None
)
# Only docker socket and addon path should be mounted
assert len(args["volumes"]) == 2
# Verify no docker config mount
for bind in args["volumes"].values():
assert bind["bind"] != "/root/.docker/config.json"

View File

@@ -2,7 +2,8 @@
# pylint: disable=protected-access
from supervisor.coresys import CoreSys
from supervisor.docker.interface import DOCKER_HUB, DockerInterface
from supervisor.docker.const import DOCKER_HUB
from supervisor.docker.interface import DockerInterface
def test_no_credentials(coresys: CoreSys, test_docker_interface: DockerInterface):