import os import requests from secret_manager import get_secret from simple_logger_handler import setup_logger dockerhub_token = get_secret(os.getenv("DOCKERHUB_TOKEN",""),"/etc/secrets/dockerhub_token") dockerhub_username = get_secret(os.getenv("DOCKERHUB_USERNAME",""),"/etc/secrets/dockerhub_username") logger = setup_logger(__name__) def login_and_get_token() -> str: """ [DockerHub] Login to Docker Hub and retrieve a JWT token. Returns: A string JWT token for authenticated requests. Returns None and logs error if login fails. """ login_url = "https://hub.docker.com/v2/users/login/" response = requests.post( login_url, json={"username": dockerhub_username, "password": dockerhub_token} ) if response.status_code == 200: token = response.json().get("token") if token: logger.debug("[DockerHub] Login successful, token obtained") return token else: logger.error("[DockerHub] Login response did not contain token") return "" else: logger.error(f"[DockerHub] Login failed: {response.status_code} - {response.text}") return "" def find_package_version_with_tag(repo: str, tag: str) -> str: """ [DockerHub] Fetch the tag ID for a repository tag. Authenticates with Docker Hub, queries the repository for the tag, returns the tag ID if found, else None. """ token = login_and_get_token() if not token: logger.error(f"[DockerHub] Cannot fetch tag '{tag}' for repo '{repo}' without token") return "" headers = {"Authorization": f"JWT {token}"} tags_url = f"https://hub.docker.com/v2/repositories/{repo}/tags/{tag}?page_size=1" response = requests.get(tags_url, headers=headers) if response.status_code != 200: logger.error(f"[DockerHub] Failed to fetch tag '{tag}' for repo '{repo}': {response.status_code} - {response.text}") return "" data = response.json() tag_id = data.get("id") if tag_id: logger.info(f"[DockerHub] Found tag '{tag}' for repo '{repo}', ID: {tag_id}") return str(tag_id) else: logger.debug(f"[DockerHub] Tag '{tag}' not found for repo '{repo}'") return "" if __name__ == "__main__": print(find_package_version_with_tag("pihole/pihole", "latest"))