1
0
mirror of https://github.com/home-assistant/supervisor.git synced 2025-12-20 02:18:59 +00:00

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>
This commit is contained in:
Stefan Agner
2025-11-27 10:27:59 +01:00
parent e7be2bfd0d
commit 5f6d7c230f
6 changed files with 40 additions and 35 deletions

View File

@@ -22,7 +22,8 @@ from ..const import (
SOCKET_DOCKER,
)
from ..coresys import CoreSys, CoreSysAttributes
from ..docker.interface import DOCKER_HUB, IMAGE_WITH_HOST, MAP_ARCH
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
from .validate import SCHEMA_BUILD_CONFIG
@@ -131,23 +132,12 @@ class AddonBuild(FileConfiguration, CoreSysAttributes):
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
base_image = self.base_image
registry = None
# Check if base image uses a custom registry
matcher = IMAGE_WITH_HOST.match(base_image)
if matcher:
if matcher.group(1) in self.sys_docker.config.registries:
registry = matcher.group(1)
# If no match, check for Docker Hub credentials
elif DOCKER_HUB in self.sys_docker.config.registries:
registry = DOCKER_HUB
registry = self.sys_docker.config.get_registry_for_image(self.base_image)
if not registry:
return None

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

@@ -10,7 +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.interface import DOCKER_HUB
from supervisor.docker.const import DOCKER_HUB
from tests.common import is_in_list

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):