| # Copyright 2026 The Bazel Authors. All rights reserved. | |
| # | |
| # Licensed under the Apache License, Version 2.0 (the "License"); | |
| # you may not use this file except in compliance with the License. | |
| # You may obtain a copy of the License at | |
| # | |
| # http://www.apache.org/licenses/LICENSE-2.0 | |
| # | |
| # Unless required by applicable law or agreed to in writing, software | |
| # distributed under the License is distributed on an "AS IS" BASIS, | |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | |
| # See the License for the specific language governing permissions and | |
| # limitations under the License. | |
| """Script to monitor SSL certificate expiration for Bazel domains.""" | |
| import datetime | |
| import os | |
| import socket | |
| import ssl | |
| import sys | |
| from typing import Any, Dict, List | |
| import certifi | |
| import yaml | |
| # Constants | |
| DEFAULT_CONFIG_PATH = os.path.join( | |
| os.path.dirname(__file__), "../config/ssl_domains.yaml" | |
| ) | |
| DEFAULT_WARNING_DAYS = 21 | |
| HTTPS_PORT = 443 | |
| SOCKET_TIMEOUT_SECONDS = 5 | |
| class SSLMonitor: | |
| """Class to manage and check SSL certificate expirations.""" | |
| def __init__(self, config_path: str = DEFAULT_CONFIG_PATH): | |
| self.config_path = config_path | |
| self.config = self._load_config() | |
| self.warning_days = self.config.get("warning_days", DEFAULT_WARNING_DAYS) | |
| self.domains = self.config.get("domains", []) | |
| self.context = ssl.create_default_context(cafile=certifi.where()) | |
| def _load_config(self) -> Dict[str, Any]: | |
| """Loads the YAML configuration file.""" | |
| if not os.path.exists(self.config_path): | |
| print(f"Error: Config file not found at {self.config_path}") | |
| sys.exit(1) | |
| try: | |
| with open(self.config_path, "r", encoding="utf-8") as f: | |
| config = yaml.safe_load(f) | |
| return config if isinstance(config, dict) else {} | |
| except yaml.YAMLError as exc: | |
| print(f"Error parsing YAML config: {exc}") | |
| sys.exit(1) | |
| def get_days_until_expiration(self, domain: str) -> int: | |
| """Connects to the domain and retrieves days until certificate expiration.""" | |
| with socket.create_connection( | |
| (domain, HTTPS_PORT), timeout=SOCKET_TIMEOUT_SECONDS | |
| ) as sock: | |
| with self.context.wrap_socket(sock, server_hostname=domain) as ssock: | |
| cert = ssock.getpeercert() | |
| if not cert: | |
| raise ValueError(f"No certificate found for {domain}") | |
| # OpenSSL always emits the notAfter month as a fixed English | |
| # abbreviation, but datetime.strptime's %b is LC_TIME-dependent and | |
| # would raise on non-English locales. ssl.cert_time_to_seconds parses | |
| # it with a hardcoded English table, so it is locale-independent. | |
| expires_ts = ssl.cert_time_to_seconds(cert["notAfter"]) | |
| expires_dt = datetime.datetime.fromtimestamp( | |
| expires_ts, datetime.timezone.utc | |
| ) | |
| delta = expires_dt - datetime.datetime.now(datetime.timezone.utc) | |
| return delta.days | |
| def check_all(self) -> List[str]: | |
| """Checks all configured domains and prints the status.""" | |
| failed_domains = [] | |
| print( | |
| f"Checking {len(self.domains)} domains (Threshold: <" | |
| f" {self.warning_days} days)...\n" | |
| ) | |
| print(f"{'DOMAIN':<35} | {'DAYS LEFT':<10} | {'STATUS'}") | |
| print("-" * 65) | |
| for domain in self.domains: | |
| try: | |
| days_left = self.get_days_until_expiration(domain) | |
| status = "✅ OK" | |
| if days_left < self.warning_days: | |
| status = "❌ EXPIRING SOON" | |
| failed_domains.append(f"{domain} ({days_left} days left)") | |
| print(f"{domain:<35} | {days_left:<10} | {status}") | |
| except (OSError, ValueError) as e: | |
| # Catch broad exceptions to ensure we try all domains | |
| error_msg = str(e) | |
| print(f"{domain:<35} | {'ERROR':<10} | 🚨 {error_msg}") | |
| failed_domains.append(f"{domain} ({error_msg})") | |
| return failed_domains | |
| def report_failures(failures: List[str]): | |
| """Reports failures to stdout and GitHub Step Summary if available.""" | |
| if not failures: | |
| print("\nAll certificates look good.") | |
| return | |
| summary = "\n".join([f"- {d}" for d in failures]) | |
| # GitHub Action specific summary | |
| if "GITHUB_STEP_SUMMARY" in os.environ: | |
| try: | |
| with open(os.environ["GITHUB_STEP_SUMMARY"], "a", encoding="utf-8") as fh: | |
| fh.write("### 🚨 SSL Certificates Expiring Soon or Failing\n") | |
| fh.write(summary + "\n") | |
| except IOError as e: | |
| print(f"Warning: Could not write to GITHUB_STEP_SUMMARY: {e}") | |
| print("\nCRITICAL ISSUES FOUND:") | |
| print(summary) | |
| sys.exit(1) | |
| def main(): | |
| monitor = SSLMonitor() | |
| if not monitor.domains: | |
| print("Error: No domains found in config.") | |
| sys.exit(1) | |
| failures = monitor.check_all() | |
| report_failures(failures) | |
| if __name__ == "__main__": | |
| main() |