service-docker-repository-query/src/dockerhub_api.py

from typing import NoReturn
from urllib.parse import quote

import requests

from repository_errors import (
	ApiAuthenticationFailure,
	OrganisationNotFoundError,
	RateLimitedError,
	RepositoryApiError,
	RepositoryNotFoundError,
	RepositoryTagNotFoundError,
)
from secret_manager import get_secret
from simple_logger_handler import setup_logger

dockerhub_token = get_secret("DOCKERHUB_TOKEN","/etc/secrets/dockerhub_token")
dockerhub_username = get_secret("DOCKERHUB_USERNAME","/etc/secrets/dockerhub_username")

logger = setup_logger(__name__)

_SERVICE = "Docker Hub"
_API = "https://hub.docker.com/v2"
_TIMEOUT = (3.05, 10)
_VERSION_FIELD = "id"

def _request(method: str, url: str, **kwargs) -> requests.Response:
	"""Send a request and turn transport problems, rate limits and 5xx into typed errors."""
	try:
		response = requests.request(method, url, timeout=_TIMEOUT, **kwargs)
	except requests.RequestException as e:
		raise RepositoryApiError(_SERVICE, f"request failed: {e}") from e

	if response.status_code == 429:
		retry_after = response.headers.get("Retry-After", "")
		raise RateLimitedError(_SERVICE, int(retry_after) if retry_after.isdigit() else None)
	if response.status_code >= 500:
		raise RepositoryApiError(_SERVICE, f"server error {response.status_code}")
	return response


def _get_access_token() -> str:
	"""
	[DockerHub] Exchange the username and personal access token for a short-lived bearer token.

	Returns:
		A string JWT token for authenticated requests.
	Raises:
		ApiAuthenticationFailure: The credentials were rejected.
		RepositoryApiError: Network problem, server error or an unexpected response.
		RateLimitedError: Docker Hub is rate limiting us.

	"""
	response = _request("POST",f"{_API}/auth/token",json={"identifier": dockerhub_username, "secret": dockerhub_token})

	if response.status_code in (401, 403):
		raise ApiAuthenticationFailure(_SERVICE, f"token request rejected ({response.status_code})")
	if response.status_code != 200:
		raise RepositoryApiError(_SERVICE, f"token request returned unexpected status {response.status_code}")

	try:
		token = response.json().get("access_token")
	except (ValueError, AttributeError) as e:
		raise RepositoryApiError(_SERVICE, "token response was not the expected JSON") from e
	if not token:
		raise RepositoryApiError(_SERVICE, "token response did not contain an access_token")

	logger.debug("[DockerHub] Access token obtained")
	return token


def _raise_not_found(repo_path: str, headers: dict[str, str], org: str, repo: str, tag: str) -> NoReturn:
	"""A 404 on the tag could mean a missing tag or a missing repository. Ask which."""
	response = _request("GET", repo_path, headers=headers)
	if response.status_code == 200:
		raise RepositoryTagNotFoundError(org, repo, tag)
	if response.status_code == 404:
		raise RepositoryNotFoundError(org, repo)
	raise RepositoryApiError(
		_SERVICE,
		f"could not tell why {org}/{repo}:{tag} was not found (status {response.status_code})",
	)

def find_package_version_with_tag(org: str, repo: str, tag: str) -> str:
	"""
	[DockerHub] Fetch the tag ID for a repository tag.

	Authenticates with Docker Hub, queries the repository for the tag,

		Raises:
		ApiAuthenticationFailure: The token request failed, or access to the repository was denied.
		RepositoryNotFoundError: The organisation/repository does not exist.
		RepositoryTagNotFoundError: The repository exists but the tag does not.
		RateLimitedError: Docker Hub is rate limiting us.
		RepositoryApiError: Network problem, server error or an unexpected response.

	"""
	repo_path = f"{_API}/namespaces/{org}/repositories/{repo}"
	url = f"{repo_path}/tags/{tag}"

	headers = {"Authorization": f"Bearer {_get_access_token()}"}
	response = _request("GET", url, headers=headers)

	if response.status_code in (401, 403):
		raise ApiAuthenticationFailure(_SERVICE, f"access to {org}/{repo} denied ({response.status_code})")
	if response.status_code == 404:
		_raise_not_found(repo_path, headers, org, repo, tag)
	if response.status_code != 200:
		raise RepositoryApiError(_SERVICE, f"unexpected status {response.status_code} for {org}/{repo}:{tag}")

	try:
		data = response.json()
	except ValueError as e:
		raise RepositoryApiError(_SERVICE, f"response for {org}/{repo}:{tag} was not valid JSON") from e

	version = data.get(_VERSION_FIELD) if isinstance(data, dict) else None
	if version is None:
		raise RepositoryApiError(_SERVICE, f"response for {org}/{repo}:{tag} had no '{_VERSION_FIELD}' field")

	logger.debug(f"[DockerHub] Found tag '{tag}' for repo '{org}/{repo}', {_VERSION_FIELD}: {version}")
	return str(version)

def _organisation_exists(org: str) -> bool:
	headers = {"Authorization": f"Bearer {_get_access_token()}"}
	response = _request("GET",f"{_API}/namespaces/{quote(org, safe='')}/repositories",params={"page_size": 1},headers=headers)

	if response.status_code == 404:
		return False
	if response.status_code in (401,403):
		raise ApiAuthenticationFailure(_SERVICE, f"listing the repositories of {org} was denied ({response.status_code})")
	if response.status_code != 200:
		raise RepositoryApiError(_SERVICE, f"unexpected status {response.status_code} when looking up {org}")

	try:
		count = response.json().get("count")
	except (ValueError,AttributeError) as e:
		raise RepositoryApiError(_SERVICE, f"response for {org} was not the expected JSON") from e
	return count > 0

def validate_repository_combination_and_return_latest_tag(org: str, repo: str, tag: str) -> str:
	"""
	[DockerHub] Check that an organisation, repository and tag exist together.

	Raises:
		OrganisationNotFoundError: The organisation has no repositories visible to us.
		RepositoryNotFoundError: The organisation exists but the repository does not.
		RepositoryTagNotFoundError: The repository exists but the tag does not.
		ApiAuthenticationFailure: The token request failed, or access was denied.
		RateLimitedError: Docker Hub is rate limiting us.
		RepositoryApiError: Network problem, server error or an unexpected response.
	"""
	try:
		return find_package_version_with_tag(org, repo, tag)
	except RepositoryNotFoundError as e:
		if not _organisation_exists(org):
			raise OrganisationNotFoundError(org) from e
		raise

if __name__ == "__main__":
	#print(find_package_version_with_tag("pihole","pihole", "latest"))
	#print(find_package_version_with_tag("library","mysql","8.4"))
	from repository_errors import RepositoryError

	try:
		print(find_package_version_with_tag("pihole","pihole", "Alatest"))
	except RepositoryError as e:
		print(f"{type(e).__name__}: {e}")