Add support to GHCR

.github/workflows/deploy.yml

@@ -40,6 +40,7 @@ jobs:
         env:
           DOCKER_USERNAME: ${{ secrets.DOCKER_USERNAME }}
           DOCKER_PASSWORD: ${{ secrets.DOCKER_PASSWORD }}
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
 
       - name: Copy repos folder for getting the extension logos
         run: |

.hooks/pre-push

@@ -20,4 +20,7 @@ pylint $(git ls-files '*.py')
 echo "Running mypy.."
 git ls-files '*.py' | xargs --max-lines=1 mypy
 
+echo "Running pytest.."
+pytest tests/ -vs
+
 exit 0

blueos_repository/docker/auth.py

@@ -8,34 +8,53 @@
 
 class DockerAuthAPI:  # pylint: disable=too-few-public-methods
     """
-    This class is used to interact with the Docker Auth API.
+    This class is used to interact with a Docker-compatible token authentication API.
+
+    Supports Docker Hub (auth.docker.io), GHCR (ghcr.io), and other OCI-compliant
+    registries that follow the Docker token authentication specification.
 
     More details in https://distribution.github.io/distribution/spec/auth/token/
     """
 
-    __api_url: str = "https://auth.docker.io"
-
-    def __init__(self, username: Optional[str] = None, password: Optional[str] = None, max_retries: int = 5) -> None:
+    def __init__(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+        self,
+        auth_url: str = "https://auth.docker.io",
+        service: str = "registry.docker.io",
+        username: Optional[str] = None,
+        password: Optional[str] = None,
+        max_retries: int = 5,
+    ) -> None:
         """
         Constructor for the DockerAuthAPI class.
 
         Args:
-            username: The username to be used in the authentication (optional)
-            password: The password to be used in the authentication (optional)
+            auth_url: Base URL for the token endpoint (e.g. "https://auth.docker.io" or "https://ghcr.io").
+            service: The ``service`` query parameter sent to the token endpoint.
+            username: The username to be used in the authentication (optional).
+            password: The password to be used in the authentication (optional).
             max_retries: The maximum number of retries to be used in case of request failure. Defaults to 5.
 
         Returns:
             None
         """
 
+        self.__api_url: str = auth_url
+        self.__service: str = service
         self.__auth_header: Optional[str] = None
 
         if username and password:
             self.__auth_header = f"Basic {base64.b64encode(f'{username}:{password}'.encode()).decode()}"
-        elif "DOCKER_USERNAME" in os.environ and "DOCKER_PASSWORD" in os.environ:
-            username = os.environ["DOCKER_USERNAME"]
-            password = os.environ["DOCKER_PASSWORD"]
-            self.__auth_header = f"Basic {base64.b64encode(f'{username}:{password}'.encode()).decode()}"
+        elif service == "registry.docker.io":
+            # Docker Hub credentials from environment
+            if "DOCKER_USERNAME" in os.environ and "DOCKER_PASSWORD" in os.environ:
+                env_user = os.environ["DOCKER_USERNAME"]
+                env_pass = os.environ["DOCKER_PASSWORD"]
+                self.__auth_header = f"Basic {base64.b64encode(f'{env_user}:{env_pass}'.encode()).decode()}"
+        elif service == "ghcr.io":
+            # GitHub Container Registry – use a GitHub token when available
+            token = os.environ.get("GITHUB_TOKEN") or os.environ.get("GH_TOKEN")
+            if token:
+                self.__auth_header = f"Basic {base64.b64encode(f'token:{token}'.encode()).decode()}"
 
         self.__retry_options = aiohttp_retry.ExponentialRetry(attempts=max_retries)
 
@@ -51,17 +70,17 @@ async def get_token(self, repo: str) -> AuthToken:
         """
 
         params = {
-            "service": "registry.docker.io",
+            "service": self.__service,
             "scope": f"repository:{repo}:pull",
         }
 
         headers = {"Authorization": self.__auth_header} if self.__auth_header else {}
 
-        auth_url = f"{self.__api_url}/token?service=registry.docker.io&scope=repository:{repo}:pull"
+        auth_url = f"{self.__api_url}/token"
         async with aiohttp_retry.RetryClient(retry_options=self.__retry_options) as session:
             async with session.get(auth_url, params=params, headers=headers) as resp:
                 if resp.status != 200:
-                    error_msg = f"Error on Docker Auth API with status {resp.status}"
+                    error_msg = f"Error on Docker Auth API ({self.__api_url}) with status {resp.status}"
                     print(error_msg)
                     raise Exception(error_msg)
 

blueos_repository/docker/image_ref.py

@@ -0,0 +1,77 @@
+import dataclasses
+
+
+@dataclasses.dataclass
+class DockerImageRef:
+    """
+    Parsed Docker image reference.
+
+    Handles both Docker Hub short references (e.g. "bluerobotics/blueos-doris")
+    and fully-qualified references with a registry hostname
+    (e.g. "ghcr.io/bluerobotics/blueos-doris").
+    """
+
+    registry: str
+    repository: str
+
+    @staticmethod
+    def parse(docker: str) -> "DockerImageRef":
+        """
+        Parse a Docker image reference string into registry and repository components.
+
+        Args:
+            docker: The docker image reference, e.g.:
+                - "bluerobotics/cockpit"                  → docker.io / bluerobotics/cockpit
+                - "ghcr.io/bluerobotics/blueos-doris"     → ghcr.io / bluerobotics/blueos-doris
+                - "docker.io/bluerobotics/cockpit"        → docker.io / bluerobotics/cockpit
+                - "registry.example.com/org/repo"         → registry.example.com / org/repo
+
+        Returns:
+            A DockerImageRef with the extracted registry and repository.
+        """
+
+        # Strip any tag or digest suffix so we only deal with the image name
+        name = docker.split("@")[0].split(":")[0]
+        parts = name.split("/")
+
+        # If the first part looks like a hostname (contains a dot or a colon,
+        # or is "localhost"), treat it as the registry.
+        if len(parts) >= 3 and ("." in parts[0] or ":" in parts[0] or parts[0] == "localhost"):
+            registry = parts[0]
+            repository = "/".join(parts[1:])
+        else:
+            registry = "docker.io"
+            repository = name
+
+        return DockerImageRef(registry=registry, repository=repository)
+
+    @property
+    def is_dockerhub(self) -> bool:
+        return self.registry in ("docker.io", "registry-1.docker.io", "index.docker.io")
+
+    @property
+    def is_ghcr(self) -> bool:
+        return self.registry == "ghcr.io"
+
+    @property
+    def registry_url(self) -> str:
+        """Base URL for the Docker Registry V2 API."""
+        if self.is_dockerhub:
+            return "https://registry-1.docker.io"
+        return f"https://{self.registry}"
+
+    @property
+    def auth_url(self) -> str:
+        """Base URL for the token authentication endpoint."""
+        if self.is_dockerhub:
+            return "https://auth.docker.io"
+        # GHCR (and most OCI registries) serve the token endpoint on the
+        # same host as the registry itself.
+        return f"https://{self.registry}"
+
+    @property
+    def auth_service(self) -> str:
+        """The ``service`` parameter sent to the token endpoint."""
+        if self.is_dockerhub:
+            return "registry.docker.io"
+        return self.registry

blueos_repository/docker/models/manifest.py

@@ -48,13 +48,13 @@ class ManifestList:
 
     Attributes:
         schemaVersion (int): This field specifies the image manifest schema version as an integer. This schema uses the version 2.
-        mediaType (str): The MIME type of the manifest list. This should be set to application/vnd.docker.distribution.manifest.list.v2+json.
         manifests (List[Manifest]): The manifests field contains a list of manifests for specific platforms.
+        mediaType (Optional[str]): The MIME type of the manifest list.
     """
 
     schemaVersion: int  # pylint: disable=invalid-name
-    mediaType: str  # pylint: disable=invalid-name
     manifests: List[Manifest]
+    mediaType: Optional[str] = None  # pylint: disable=invalid-name
 
 
 @dataclasses.dataclass
@@ -98,15 +98,15 @@ class ImageManifest:
 
     Attributes:
         schemaVersion (int): The image manifest schema version as an integer, version 2 is expected.
-        mediaType (str): MIME type of the manifest, expected to be application/vnd.docker.distribution.manifest.v2+json.
+        mediaType (Optional[str]): MIME type of the manifest. Optional per OCI spec.
         config (ConfigObject): The configuration object for a container by digest.
         layers (List[Layer]): Ordered list of layers starting from the base image.
     """
 
     schemaVersion: int  # pylint: disable=invalid-name
-    mediaType: str  # pylint: disable=invalid-name
     config: ConfigReference
     layers: List[ManifestLayer]
+    mediaType: Optional[str] = None  # pylint: disable=invalid-name  # Optional per OCI spec
 
 
 @dataclasses.dataclass

blueos_repository/docker/registry.py

@@ -1,4 +1,4 @@
-from typing import Any, Optional
+from typing import Any, List, Optional
 
 import aiohttp_retry
 from dataclass_wizard import fromdict
@@ -9,18 +9,26 @@
 from docker.models.manifest import ImageManifest, ManifestFetch, ManifestList
 
 
-class DockerRegistry:
+class DockerRegistry:  # pylint: disable=too-many-instance-attributes
     """
-    This class is used to interact with the Docker Registry API.
+    This class is used to interact with a Docker-compatible Registry V2 API.
+
+    Supports Docker Hub (registry-1.docker.io), GHCR (ghcr.io), and other
+    OCI-compliant registries.
 
     More details in https://distribution.github.io/distribution/spec/api/
     """
 
-    __api_base_url: str = "https://registry-1.docker.io"
-    __api_version: str = "v2"
-    __api_url: str = f"{__api_base_url}/{__api_version}"
-
-    __token: Optional[AuthToken] = None
+    # Media types accepted when fetching manifests.  We request every format
+    # the codebase can handle so the registry returns the best match.
+    __manifest_accept: str = ",".join(
+        [
+            "application/vnd.docker.distribution.manifest.v2+json",
+            "application/vnd.docker.distribution.manifest.list.v2+json",
+            "application/vnd.oci.image.manifest.v1+json",
+            "application/vnd.oci.image.index.v1+json",
+        ]
+    )
 
     @staticmethod
     def from_preview() -> "DockerRegistry":
@@ -33,18 +41,35 @@ def from_preview() -> "DockerRegistry":
 
         return DockerRegistry("ratelimitpreview/test")
 
-    def __init__(self, repository: str, max_retries: int = 5) -> None:
+    def __init__(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+        self,
+        repository: str,
+        registry_url: str = "https://registry-1.docker.io",
+        auth_url: str = "https://auth.docker.io",
+        auth_service: str = "registry.docker.io",
+        max_retries: int = 5,
+    ) -> None:
         """
-        Constructor for the DockerHubAPI class.
+        Constructor for the DockerRegistry class.
 
         Args:
-            repository: Repository that this registry class will operate on
+            repository: Repository that this registry class will operate on.
+            registry_url: Base URL for the registry (e.g. "https://registry-1.docker.io" or "https://ghcr.io").
+            auth_url: Base URL for the token authentication endpoint.
+            auth_service: The ``service`` parameter used for token requests.
             max_retries: The maximum number of retries to be used in case of request failure. Defaults to 5.
 
         Returns:
             None
         """
+
         self.repository = repository
+        self.__api_base_url: str = registry_url
+        self.__auth_url: str = auth_url
+        self.__auth_service: str = auth_service
+        self.__api_version: str = "v2"
+        self.__api_url: str = f"{self.__api_base_url}/{self.__api_version}"
+        self.__token: Optional[AuthToken] = None
         self.__retry_options = aiohttp_retry.ExponentialRetry(attempts=max_retries)
 
     async def __check_token(self) -> None:
@@ -53,7 +78,7 @@ async def __check_token(self) -> None:
         """
 
         if not self.__token or self.__token.is_expired:
-            auth = DockerAuthAPI()
+            auth = DockerAuthAPI(auth_url=self.__auth_url, service=self.__auth_service)
             self.__token = await auth.get_token(self.repository)
 
     async def __raise_pretty(self, resp: Any) -> None:
@@ -81,12 +106,12 @@ async def __raise_pretty(self, resp: Any) -> None:
 
     async def get(self, route: str, max_retries: Optional[int] = None, **kwargs: Any) -> Any:
         """
-        Make a GET request to the Docker Hub API.
+        Make a GET request to the Docker Registry V2 API.
 
         Args:
             route: The route to be used in the request.
-            params: The parameters to be used in the request.
             max_retries: The maximum number of retries to be used in case of request failure. Defaults to None.
+            **kwargs: Additional keyword arguments passed to the HTTP client (e.g. headers, params).
 
         Returns:
             The response from the request parsed as json.
@@ -121,14 +146,20 @@ async def get_manifest(self, tag_or_digest: str) -> ManifestFetch:
         route = f"{self.repository}/manifests/{tag_or_digest}"
         header = {
             "Authorization": f"Bearer {self.__token.token if self.__token else ''}",
-            "Accept": "application/vnd.docker.distribution.manifest.v2+json,application/vnd.oci.image.manifest.v1+json,application/vnd.oci.image.index.v1+json",
+            "Accept": self.__manifest_accept,
         }
 
         manifest = await self.get(route, headers=header)
 
         if "config" in manifest:
             return ManifestFetch(manifest=fromdict(ImageManifest, manifest))
 
+        # OCI indexes (e.g. from GHCR) may contain non-image entries such as
+        # attestation manifests that lack a "platform" field.  Filter them out
+        # so the typed Manifest dataclass can require platform unconditionally.
+        if "manifests" in manifest:
+            manifest["manifests"] = [m for m in manifest["manifests"] if "platform" in m]
+
         return ManifestFetch(manifest=fromdict(ManifestList, manifest))
 
     async def get_manifest_blob(self, digest: str) -> Blob:
@@ -147,13 +178,32 @@ async def get_manifest_blob(self, digest: str) -> Blob:
         route = f"{self.repository}/blobs/{digest}"
         header = {
             "Authorization": f"Bearer {self.__token.token if self.__token else ''}",
-            "Accept": "application/vnd.docker.distribution.manifest.v2+json,application/vnd.oci.image.manifest.v1+json,application/vnd.oci.image.index.v1+json",
+            "Accept": "application/vnd.oci.image.config.v1+json,application/json",
         }
 
         blob = await self.get(route, headers=header)
 
         return fromdict(Blob, blob)
 
+    async def list_tags(self) -> List[str]:
+        """
+        List all tags for the repository using the standard V2 tag listing API.
+
+        This is the registry-agnostic way to discover tags; it works on any
+        OCI-compliant registry (Docker Hub, GHCR, Quay, etc.).
+
+        Returns:
+            A list of tag name strings.
+        """
+
+        await self.__check_token()
+
+        route = f"{self.repository}/tags/list"
+        header = {"Authorization": f"Bearer {self.__token.token if self.__token else ''}"}
+
+        data = await self.get(route, headers=header)
+        return list(data.get("tags", []) or [])
+
     async def get_rate_limit(self) -> RateLimit:
 
         await self.__check_token()