import os import sys import time import logging import requests from typing import Optional # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.StreamHandler(sys.stdout) ] ) logger = logging.getLogger(__name__) # Environment variables DOCKER_REPO_URL = os.getenv('DOCKER_REPO_URL') DOCKER_IMAGE = os.getenv('DOCKER_IMAGE') WEBHOOK_URL = os.getenv('WEBHOOK_URL') WEBHOOK_TOKEN = os.getenv('WEBHOOK_TOKEN') # SHA storage file SHA_FILE = '/app/sha.txt' # Check interval in seconds (default: 300 = 5 minutes) CHECK_INTERVAL = int(os.getenv('CHECK_INTERVAL', '300')) def validate_env_vars(): """Validate that all required environment variables are set.""" missing = [] if not DOCKER_REPO_URL: missing.append('DOCKER_REPO_URL') if not DOCKER_IMAGE: missing.append('DOCKER_IMAGE') if not WEBHOOK_URL: missing.append('WEBHOOK_URL') if not WEBHOOK_TOKEN: missing.append('WEBHOOK_TOKEN') if missing: logger.error(f"Missing required environment variables: {', '.join(missing)}") sys.exit(1) def get_image_sha(repo_url: str, image: str) -> Optional[str]: """ Get the SHA256 digest of the Docker image from the registry. Args: repo_url: Docker registry URL (e.g., docker.io, ghcr.io, or custom registry) image: Full image name with tag (e.g., 'username/image:tag' or 'username/image:latest') Returns: SHA256 digest string or None if failed """ try: # Parse image name and tag if ':' in image: image_name, tag = image.rsplit(':', 1) else: image_name = image tag = 'latest' # Normalize repo URL (remove https://, http://, trailing /) repo_url_normalized = repo_url.rstrip('/').replace('https://', '').replace('http://', '') # Construct manifest URL # For Docker Hub: https://registry-1.docker.io/v2/{namespace}/{image}/manifests/{tag} # For other registries: https://{repo_url}/v2/{image}/manifests/{tag} if repo_url_normalized in ('docker.io', 'hub.docker.com', ''): # Docker Hub uses registry-1.docker.io registry_url = 'https://registry-1.docker.io' # For Docker Hub, image_name should be 'username/image' or 'library/image' for official images # If no namespace, assume it's an official image (library namespace) if '/' not in image_name: image_name = f'library/{image_name}' manifest_url = f"{registry_url}/v2/{image_name}/manifests/{tag}" else: # Other registries (ghcr.io, quay.io, custom registries) registry_url = f"https://{repo_url_normalized}" manifest_url = f"{registry_url}/v2/{image_name}/manifests/{tag}" logger.debug(f"Fetching manifest from: {manifest_url}") # Request manifest with Accept header for schema v2 headers = { 'Accept': 'application/vnd.docker.distribution.manifest.v2+json, application/vnd.docker.distribution.manifest.list.v2+json' } response = requests.get(manifest_url, headers=headers, timeout=30) response.raise_for_status() # Get SHA from Docker-Content-Digest header (preferred) sha = response.headers.get('Docker-Content-Digest') if not sha: # Fallback: try to get from response if it's a manifest list try: manifest = response.json() # For manifest lists, get the digest from the first platform-specific manifest if manifest.get('mediaType') == 'application/vnd.docker.distribution.manifest.list.v2+json': if 'manifests' in manifest and len(manifest['manifests']) > 0: sha = manifest['manifests'][0].get('digest') elif 'config' in manifest and 'digest' in manifest['config']: sha = manifest['config']['digest'] elif 'digest' in manifest: sha = manifest['digest'] except Exception as e: logger.debug(f"Could not parse manifest JSON: {e}") if sha: logger.info(f"Current SHA for {image}: {sha[:16]}...") return sha else: logger.warning(f"Could not extract SHA from manifest response") return None except requests.exceptions.HTTPError as e: if e.response.status_code == 404: logger.error(f"Image not found: {image} (tag: {tag})") elif e.response.status_code == 401: logger.error(f"Authentication required for {repo_url}. Private images may require authentication.") else: logger.error(f"HTTP error fetching image SHA: {e}") return None except requests.exceptions.RequestException as e: logger.error(f"Failed to fetch image SHA: {e}") return None except Exception as e: logger.error(f"Unexpected error fetching image SHA: {e}") return None def load_stored_sha() -> Optional[str]: """Load the previously stored SHA from file.""" try: if os.path.exists(SHA_FILE): with open(SHA_FILE, 'r') as f: sha = f.read().strip() if sha: logger.debug(f"Loaded stored SHA: {sha}") return sha except Exception as e: logger.warning(f"Failed to load stored SHA: {e}") return None def save_sha(sha: str): """Save the SHA to file.""" try: with open(SHA_FILE, 'w') as f: f.write(sha) logger.debug(f"Saved SHA: {sha}") except Exception as e: logger.error(f"Failed to save SHA: {e}") def trigger_webhook(url: str, token: str) -> bool: """ Trigger the webhook with bearer token authentication. Args: url: Webhook URL token: Bearer token Returns: True if successful, False otherwise """ try: headers = { 'Authorization': f'Bearer {token}', 'Content-Type': 'application/json' } # Send POST request (you can modify payload if needed) payload = { 'event': 'image_updated', 'image': DOCKER_IMAGE, 'repo_url': DOCKER_REPO_URL } logger.info(f"Triggering webhook: {url}") response = requests.post(url, json=payload, headers=headers, timeout=30) response.raise_for_status() logger.info(f"Webhook triggered successfully (status: {response.status_code})") return True except requests.exceptions.RequestException as e: logger.error(f"Failed to trigger webhook: {e}") return False except Exception as e: logger.error(f"Unexpected error triggering webhook: {e}") return False def main(): """Main loop to check for image updates.""" logger.info("Starting Docker Webhook Updater") logger.info(f"Monitoring: {DOCKER_REPO_URL}/{DOCKER_IMAGE}") logger.info(f"Webhook URL: {WEBHOOK_URL}") logger.info(f"Check interval: {CHECK_INTERVAL} seconds") validate_env_vars() # Load initial SHA stored_sha = load_stored_sha() if stored_sha: logger.info(f"Initial stored SHA: {stored_sha}") else: logger.info("No stored SHA found. Will trigger webhook on first check if image exists.") while True: try: # Get current SHA from registry current_sha = get_image_sha(DOCKER_REPO_URL, DOCKER_IMAGE) if current_sha: # Compare with stored SHA if stored_sha and current_sha != stored_sha: logger.info(f"Image updated! Old SHA: {stored_sha[:16]}..., New SHA: {current_sha[:16]}...") # Trigger webhook if trigger_webhook(WEBHOOK_URL, WEBHOOK_TOKEN): # Save new SHA only if webhook was successful save_sha(current_sha) stored_sha = current_sha else: logger.warning("Webhook failed. SHA not updated. Will retry on next check.") elif not stored_sha: # First run - just store the SHA without triggering webhook logger.info(f"First run: storing initial SHA: {current_sha[:16]}...") save_sha(current_sha) stored_sha = current_sha else: logger.debug(f"No update detected. SHA: {current_sha[:16]}...") else: logger.warning("Could not fetch image SHA. Will retry on next check.") # Wait before next check logger.debug(f"Waiting {CHECK_INTERVAL} seconds before next check...") time.sleep(CHECK_INTERVAL) except KeyboardInterrupt: logger.info("Received interrupt signal. Shutting down...") break except Exception as e: logger.error(f"Unexpected error in main loop: {e}") logger.info(f"Waiting {CHECK_INTERVAL} seconds before retry...") time.sleep(CHECK_INTERVAL) if __name__ == "__main__": main()