|
| 1 | +""" |
| 2 | +Version Check Endpoint |
| 3 | +
|
| 4 | +Provides current version info and checks GitHub for available updates. |
| 5 | +No authentication required — version info is not sensitive. |
| 6 | +""" |
| 7 | + |
| 8 | +import logging |
| 9 | +import os |
| 10 | +import time |
| 11 | +from typing import Optional |
| 12 | + |
| 13 | +import httpx |
| 14 | +from fastapi import APIRouter |
| 15 | +from pydantic import BaseModel |
| 16 | + |
| 17 | +logger = logging.getLogger(__name__) |
| 18 | + |
| 19 | +router = APIRouter(prefix="/vyos/version", tags=["version"]) |
| 20 | + |
| 21 | +GITHUB_REPO = "Community-VyProjects/VyManager" |
| 22 | +CACHE_TTL_SECONDS = 3600 # 1 hour |
| 23 | + |
| 24 | + |
| 25 | +class VersionCheckResponse(BaseModel): |
| 26 | + current_version: str |
| 27 | + latest_version: Optional[str] = None |
| 28 | + update_available: bool = False |
| 29 | + release_url: Optional[str] = None |
| 30 | + published_at: Optional[str] = None |
| 31 | + environment: str |
| 32 | + |
| 33 | + |
| 34 | +# Simple in-memory cache |
| 35 | +_cache: dict[str, object] = {} |
| 36 | +_cache_time: float = 0.0 |
| 37 | + |
| 38 | + |
| 39 | +async def _fetch_latest_release() -> Optional[dict[str, str]]: |
| 40 | + """Fetch the latest release from GitHub API with 1-hour caching.""" |
| 41 | + global _cache, _cache_time |
| 42 | + |
| 43 | + now = time.time() |
| 44 | + if _cache and (now - _cache_time) < CACHE_TTL_SECONDS: |
| 45 | + return _cache # type: ignore[return-value] |
| 46 | + |
| 47 | + try: |
| 48 | + async with httpx.AsyncClient(timeout=10.0) as client: |
| 49 | + resp = await client.get( |
| 50 | + f"https://api.github.com/repos/{GITHUB_REPO}/releases/latest", |
| 51 | + headers={"Accept": "application/vnd.github+json"}, |
| 52 | + ) |
| 53 | + if resp.status_code == 200: |
| 54 | + data = resp.json() |
| 55 | + result = { |
| 56 | + "tag_name": data.get("tag_name", ""), |
| 57 | + "html_url": data.get("html_url", ""), |
| 58 | + "published_at": data.get("published_at", ""), |
| 59 | + } |
| 60 | + _cache = result |
| 61 | + _cache_time = now |
| 62 | + return result |
| 63 | + logger.warning("GitHub API returned status %d", resp.status_code) |
| 64 | + return None |
| 65 | + except Exception: |
| 66 | + logger.exception("Failed to fetch latest release from GitHub") |
| 67 | + return None |
| 68 | + |
| 69 | + |
| 70 | +def _parse_version(version_str: str) -> Optional[tuple[int, ...]]: |
| 71 | + """Parse a version string like '1.0.0-beta.1' into comparable tuples. |
| 72 | +
|
| 73 | + Returns a tuple suitable for comparison. Beta versions sort below their |
| 74 | + release counterpart (e.g. 1.0.0-beta.1 < 1.0.0). |
| 75 | + """ |
| 76 | + clean = version_str.lstrip("v") |
| 77 | + try: |
| 78 | + # Split on hyphen: "1.0.0-beta.1" -> ["1.0.0", "beta.1"] |
| 79 | + parts = clean.split("-", 1) |
| 80 | + core = tuple(int(x) for x in parts[0].split(".")) |
| 81 | + |
| 82 | + if len(parts) == 1: |
| 83 | + # Release version: use a high pre-release marker so it sorts above betas |
| 84 | + return core + (1, 0) |
| 85 | + |
| 86 | + # Pre-release: extract the numeric suffix |
| 87 | + pre = parts[1] # e.g. "beta.1" |
| 88 | + pre_parts = pre.split(".") |
| 89 | + pre_num = int(pre_parts[-1]) if pre_parts[-1].isdigit() else 0 |
| 90 | + # 0 in the fourth position means pre-release (sorts below release's 1) |
| 91 | + return core + (0, pre_num) |
| 92 | + except (ValueError, IndexError): |
| 93 | + return None |
| 94 | + |
| 95 | + |
| 96 | +def _is_update_available(current: str, latest: str) -> bool: |
| 97 | + """Compare version strings. Returns True if latest > current.""" |
| 98 | + if current == "dev": |
| 99 | + return False |
| 100 | + |
| 101 | + current_tuple = _parse_version(current) |
| 102 | + latest_tuple = _parse_version(latest) |
| 103 | + |
| 104 | + if current_tuple is None or latest_tuple is None: |
| 105 | + return False |
| 106 | + |
| 107 | + return latest_tuple > current_tuple |
| 108 | + |
| 109 | + |
| 110 | +@router.get("/check", response_model=VersionCheckResponse) |
| 111 | +async def check_version() -> VersionCheckResponse: |
| 112 | + """Check for available updates against the latest GitHub release.""" |
| 113 | + current_version = os.environ.get("VYMANAGER_VERSION", "dev") |
| 114 | + environment = os.environ.get("VYMANAGER_ENV", "dev") |
| 115 | + |
| 116 | + release = await _fetch_latest_release() |
| 117 | + |
| 118 | + if not release: |
| 119 | + return VersionCheckResponse( |
| 120 | + current_version=current_version, |
| 121 | + update_available=False, |
| 122 | + environment=environment, |
| 123 | + ) |
| 124 | + |
| 125 | + tag = release["tag_name"] |
| 126 | + latest_version = tag.lstrip("v") |
| 127 | + |
| 128 | + return VersionCheckResponse( |
| 129 | + current_version=current_version, |
| 130 | + latest_version=latest_version, |
| 131 | + update_available=_is_update_available(current_version, latest_version), |
| 132 | + release_url=release["html_url"], |
| 133 | + published_at=release["published_at"], |
| 134 | + environment=environment, |
| 135 | + ) |
0 commit comments