258 lines
9.1 KiB
Python
258 lines
9.1 KiB
Python
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()
|