From 7181cdbd1b58081607ba9d8295538293c9227616 Mon Sep 17 00:00:00 2001 From: Florian Date: Tue, 3 Nov 2026 10:40:16 +0100 Subject: [PATCH 1/2] Overhauled API calls, it now raises meaningful error messagess that can be understood by an upstream client --- src/dockerhub_api.py | 178 ++++++++++++++++++++++++++++++--------- src/github_api.py | 116 ++++++++++++++++++++----- src/repository_errors.py | 92 ++++++++++++++++++++ 3 files changed, 325 insertions(+), 61 deletions(-) create mode 100644 src/repository_errors.py diff --git a/src/dockerhub_api.py b/src/dockerhub_api.py index 21bbb40..501b507 100644 --- a/src/dockerhub_api.py +++ b/src/dockerhub_api.py @@ -1,6 +1,16 @@ +from typing import NoReturn +from urllib.parse import quote import requests +from repository_errors import ( + ApiAuthenticationFailure, + OrganisationNotFoundError, + RateLimitedError, + RepositoryApiError, + RepositoryNotFoundError, + RepositoryTagNotFoundError, +) from secret_manager import get_secret from simple_logger_handler import setup_logger @@ -9,63 +19,149 @@ dockerhub_username = get_secret("DOCKERHUB_USERNAME","/etc/secrets/dockerhub_use logger = setup_logger(__name__) -def login_and_get_token() -> str: +_SERVICE = "Docker Hub" +_API = "https://hub.docker.com/v2" +_TIMEOUT = (3.05, 10) +_VERSION_FIELD = "id" + +def _request(method: str, url: str, **kwargs) -> requests.Response: + """Send a request and turn transport problems, rate limits and 5xx into typed errors.""" + try: + response = requests.request(method, url, timeout=_TIMEOUT, **kwargs) + except requests.RequestException as e: + raise RepositoryApiError(_SERVICE, f"request failed: {e}") from e + + if response.status_code == 429: + retry_after = response.headers.get("Retry-After", "") + raise RateLimitedError(_SERVICE, int(retry_after) if retry_after.isdigit() else None) + if response.status_code >= 500: + raise RepositoryApiError(_SERVICE, f"server error {response.status_code}") + return response + + +def _get_access_token() -> str: """ - [DockerHub] Login to Docker Hub and retrieve a JWT token. + [DockerHub] Exchange the username and personal access token for a short-lived bearer token. Returns: A string JWT token for authenticated requests. - Returns None and logs error if login fails. + Raises: + ApiAuthenticationFailure: The credentials were rejected. + RepositoryApiError: Network problem, server error or an unexpected response. + RateLimitedError: Docker Hub is rate limiting us. + """ - login_url = "https://hub.docker.com/v2/users/login/" - response = requests.post( - login_url, - json={"username": dockerhub_username, "password": dockerhub_token} + response = _request("POST",f"{_API}/auth/token",json={"identifier": dockerhub_username, "secret": dockerhub_token}) + + if response.status_code in (401, 403): + raise ApiAuthenticationFailure(_SERVICE, f"token request rejected ({response.status_code})") + if response.status_code != 200: + raise RepositoryApiError(_SERVICE, f"token request returned unexpected status {response.status_code}") + + try: + token = response.json().get("access_token") + except (ValueError, AttributeError) as e: + raise RepositoryApiError(_SERVICE, "token response was not the expected JSON") from e + if not token: + raise RepositoryApiError(_SERVICE, "token response did not contain an access_token") + + logger.debug("[DockerHub] Access token obtained") + return token + + +def _raise_not_found(repo_path: str, headers: dict[str, str], org: str, repo: str, tag: str) -> NoReturn: + """A 404 on the tag could mean a missing tag or a missing repository. Ask which.""" + response = _request("GET", repo_path, headers=headers) + if response.status_code == 200: + raise RepositoryTagNotFoundError(org, repo, tag) + if response.status_code == 404: + raise RepositoryNotFoundError(org, repo) + raise RepositoryApiError( + _SERVICE, + f"could not tell why {org}/{repo}:{tag} was not found (status {response.status_code})", ) - 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: +def find_package_version_with_tag(org: str, 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. + + Raises: + ApiAuthenticationFailure: The token request failed, or access to the repository was denied. + RepositoryNotFoundError: The organisation/repository does not exist. + RepositoryTagNotFoundError: The repository exists but the tag does not. + RateLimitedError: Docker Hub is rate limiting us. + RepositoryApiError: Network problem, server error or an unexpected response. + """ - token = login_and_get_token() - if not token: - logger.error(f"[DockerHub] Cannot fetch tag '{tag}' for repo '{repo}' without token") - return "" + repo_path = f"{_API}/namespaces/{org}/repositories/{repo}" + url = f"{repo_path}/tags/{tag}" - headers = {"Authorization": f"JWT {token}"} - tags_url = f"https://hub.docker.com/v2/repositories/{repo}/tags/{tag}?page_size=1" + headers = {"Authorization": f"Bearer {_get_access_token()}"} + response = _request("GET", url, headers=headers) - response = requests.get(tags_url, headers=headers) + if response.status_code in (401, 403): + raise ApiAuthenticationFailure(_SERVICE, f"access to {org}/{repo} denied ({response.status_code})") + if response.status_code == 404: + _raise_not_found(repo_path, headers, org, repo, tag) if response.status_code != 200: - logger.error(f"[DockerHub] Failed to fetch tag '{tag}' for repo '{repo}': {response.status_code} - {response.text}") - return "" + raise RepositoryApiError(_SERVICE, f"unexpected status {response.status_code} for {org}/{repo}:{tag}") - 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 "" + try: + data = response.json() + except ValueError as e: + raise RepositoryApiError(_SERVICE, f"response for {org}/{repo}:{tag} was not valid JSON") from e + version = data.get(_VERSION_FIELD) if isinstance(data, dict) else None + if version is None: + raise RepositoryApiError(_SERVICE, f"response for {org}/{repo}:{tag} had no '{_VERSION_FIELD}' field") + + logger.debug(f"[DockerHub] Found tag '{tag}' for repo '{org}/{repo}', {_VERSION_FIELD}: {version}") + return str(version) + +def _organisation_exists(org: str) -> bool: + headers = {"Authorization": f"Bearer {_get_access_token()}"} + response = _request("GET",f"{_API}/namespaces/{quote(org, safe='')}/repositories",params={"page_size": 1},headers=headers) + + if response.status_code == 404: + return False + if response.status_code in (401,403): + raise ApiAuthenticationFailure(_SERVICE, f"listing the repositories of {org} was denied ({response.status_code})") + if response.status_code != 200: + raise RepositoryApiError(_SERVICE, f"unexpected status {response.status_code} when looking up {org}") + + try: + count = response.json().get("count") + except (ValueError,AttributeError) as e: + raise RepositoryApiError(_SERVICE, f"response for {org} was not the expected JSON") from e + return count > 0 + +def validate_repository_combination_and_return_latest_tag(org: str, repo: str, tag: str) -> str: + """ + [DockerHub] Check that an organisation, repository and tag exist together. + + Raises: + OrganisationNotFoundError: The organisation has no repositories visible to us. + RepositoryNotFoundError: The organisation exists but the repository does not. + RepositoryTagNotFoundError: The repository exists but the tag does not. + ApiAuthenticationFailure: The token request failed, or access was denied. + RateLimitedError: Docker Hub is rate limiting us. + RepositoryApiError: Network problem, server error or an unexpected response. + """ + try: + return find_package_version_with_tag(org, repo, tag) + except RepositoryNotFoundError as e: + if not _organisation_exists(org): + raise OrganisationNotFoundError(org) from e + raise if __name__ == "__main__": - #print(find_package_version_with_tag("pihole/pihole", "latest")) - print(find_package_version_with_tag("library/mysql","8.4")) \ No newline at end of file + #print(find_package_version_with_tag("pihole","pihole", "latest")) + #print(find_package_version_with_tag("library","mysql","8.4")) + from repository_errors import RepositoryError + + try: + print(find_package_version_with_tag("pihole","pihole", "Alatest")) + except RepositoryError as e: + print(f"{type(e).__name__}: {e}") diff --git a/src/github_api.py b/src/github_api.py index c2806d2..e2cc466 100644 --- a/src/github_api.py +++ b/src/github_api.py @@ -1,50 +1,126 @@ +from urllib.parse import quote import requests +from repository_errors import ( + ApiAuthenticationFailure, + OrganisationNotFoundError, + RateLimitedError, + RepositoryApiError, + RepositoryNotFoundError, + RepositoryTagNotFoundError, +) from secret_manager import get_secret from simple_logger_handler import setup_logger github_token = get_secret("GITHUB_TOKEN","/etc/secrets/github_token") logger = setup_logger(__name__) +_SERVICE = "GitHub" +_API = "https://api.github.com" +_TIMEOUT = (3.05, 10) +_PER_PAGE = 100 +_HEADERS = { + "Authorization": f"Bearer {github_token}", + "Accept": "application/vnd.github+json", +} + +def _request(url: str, **kwargs) -> requests.Response: + """GET a URL and turn transport problems, rate limits and 5xx into typed errors.""" + try: + response = requests.get(url, headers=_HEADERS, timeout=_TIMEOUT, **kwargs) + except requests.RequestException as e: + raise RepositoryApiError(_SERVICE, f"request failed: {e}") from e + + # GitHub signals rate limits as 429, or as 403 with an exhausted quota or a Retry-After header + rate_limited = response.status_code == 429 or ( + response.status_code == 403 + and (response.headers.get("X-RateLimit-Remaining") == "0" or "Retry-After" in response.headers) + ) + if rate_limited: + retry_after = response.headers.get("Retry-After", "") + raise RateLimitedError(_SERVICE, int(retry_after) if retry_after.isdigit() else None) + if response.status_code >= 500: + raise RepositoryApiError(_SERVICE, f"server error {response.status_code}") + return response + def find_package_version_with_tag(org: str, package: str, target_tag: str) -> str: """ - [GitHub] Search for a container package version with the specified tag. + [GitHub] Return the ID of the container package version that carries the tag. - Iterates through GitHub Package Registry pages for a given org and package, - returns the version ID matching the target_tag, or None if not found. + Pages through the versions of the package, newest first, until the tag is found. + + Raises: + ApiAuthenticationFailure: The token was rejected or lacks access to the package. + RepositoryNotFoundError: The package does not exist, or the token cannot see it. + RepositoryTagNotFoundError: Every page was searched and none carries the tag. + RateLimitedError: GitHub is rate limiting us. + RepositoryApiError: Network problem, server error or an unexpected response. """ - headers = { - "Authorization": f"Bearer {github_token}", - "Accept": "application/vnd.github+json" - } + url = f"{_API}/orgs/{quote(org, safe='')}/packages/container/{quote(package, safe='')}/versions" page = 1 - per_page = 100 - while True: - url = f"https://api.github.com/orgs/{org}/packages/container/{package}/versions" - params = {"per_page": per_page, "page": page} + response = _request(url, params={"per_page": _PER_PAGE, "page": page}) - response = requests.get(url, headers=headers, params=params) + if response.status_code in (401, 403): + raise ApiAuthenticationFailure(_SERVICE, f"access to {org}/{package} denied ({response.status_code})") + if response.status_code == 404: + raise RepositoryNotFoundError(org, package) if response.status_code != 200: - logger.error(f"[GitHub] Error {response.status_code} fetching {package}: {response.text}") - return "" + raise RepositoryApiError(_SERVICE, f"unexpected status {response.status_code} for {org}/{package} (page {page})") - versions = response.json() - if not versions: - logger.debug(f"[GitHub] Reached end of pages — tag '{target_tag}' not found for {package}.") - return "" + try: + versions = response.json() + except ValueError as e: + raise RepositoryApiError(_SERVICE, f"response for {org}/{package} was not valid JSON") from e + if not isinstance(versions, list): + raise RepositoryApiError(_SERVICE, f"response for {org}/{package} was not a list of versions") for version in versions: tags = version.get("metadata", {}).get("container", {}).get("tags", []) if target_tag in tags: - logger.info(f"[GitHub] Found tag '{target_tag}' for {package} on page {page}, version ID: {version['id']}") + version_id = version.get("id") + if version_id is None: + raise RepositoryApiError(_SERVICE, f"version of {org}/{package}:{target_tag} had no id") + logger.info(f"[GitHub] Found tag '{target_tag}' for {package} on page {page}, version ID: {version_id}") return str(version["id"]) + if len(versions) < _PER_PAGE: + raise RepositoryTagNotFoundError(org, package, target_tag) page += 1 logger.debug(f"[GitHub] Tag '{target_tag}' not found on page {page-1}, moving to page {page}") +def _organisation_exists(org: str) -> bool: + """Ask GitHub whether the organisation exists.""" + response = _request(f"{_API}/orgs/{quote(org, safe='')}") + if response.status_code == 200: + return True + if response.status_code == 404: + return False + if response.status_code in (401, 403): + raise ApiAuthenticationFailure(_SERVICE, f"looking up organisation {org} was denied ({response.status_code})") + raise RepositoryApiError(_SERVICE, f"unexpected status {response.status_code} when looking up {org}") + + +def validate_repository_combination_and_return_latest_tag(org: str, repo: str, tag: str) -> str: + """ + [GitHub] Check that an organisation, package and tag exist together, and return the tag's version ID. + + Raises: + OrganisationNotFoundError: The organisation does not exist. + RepositoryNotFoundError: The organisation exists but the package does not (or is not visible to the token). + RepositoryTagNotFoundError: The package exists but no version carries the tag. + ApiAuthenticationFailure: The token was rejected or lacks access. + RateLimitedError: GitHub is rate limiting us. + RepositoryApiError: Network problem, server error or an unexpected response. + """ + try: + return find_package_version_with_tag(org, repo, tag) + except RepositoryNotFoundError as e: + if not _organisation_exists(org): + raise OrganisationNotFoundError(org) from e + raise if __name__ == "__main__": - find_package_version_with_tag("Suwayomi", "tachidesk", "stable") \ No newline at end of file + validate_repository_combination_and_return_latest_tag("Suwayomis", "tachidesk", "stable") \ No newline at end of file diff --git a/src/repository_errors.py b/src/repository_errors.py new file mode 100644 index 0000000..a986dd7 --- /dev/null +++ b/src/repository_errors.py @@ -0,0 +1,92 @@ +class RepositoryError(Exception): + status_code = 502 + code = "drq_upstream_error" + client_visible = False + + @property + def client_message(self) -> str: + return str(self) if self.client_visible else "The repository could not be checked right now, please try again later" + +class InvalidOrganisationError(RepositoryError, ValueError): + status_code = 422 + code = "drq_invalid_organisation" + client_visible = True + def __init__(self, org:str): + super().__init__(f"Invalid organisation name: {org!r}") + +class InvalidRepositoryError(RepositoryError, ValueError): + status_code = 422 + code = "drq_invalid_repository" + client_visible = True + def __init__(self, repo:str): + super().__init__(f"Invalid repository name: {repo!r}") + +class InvalidTagError(RepositoryError, ValueError): + status_code = 422 + code = "drq_invalid_tag" + client_visible = True + def __init__(self, tag:str): + super().__init__(f"Invalid tag: {tag!r}") + +class ApiAuthenticationFailure(RepositoryError): + """The registry rejected our credentials or token. + + Not specific to one repository: every lookup against the same registry + will normally fail the same way. + """ + + def __init__(self, service: str, detail: str = ""): + message = f"{service} authentication failed" + super().__init__(f"{message}: {detail}" if detail else message) + + +class RepositoryNotFoundError(RepositoryError): + """The organisation or repository does not exist (or is not visible to us).""" + status_code = 422 + code = "drq_repository_not_found" + client_visible = True + def __init__(self, org: str, repo: str): + self.org = org + self.repo = repo + super().__init__(f"Repository not found: {org}/{repo}") + + +class RepositoryTagNotFoundError(RepositoryError): + """The repository exists but the tag does not. It may have been deleted.""" + status_code = 422 + code = "drq_tag_not_found" + client_visible = True + def __init__(self, org: str, repo: str, tag: str): + self.org = org + self.repo = repo + self.tag = tag + super().__init__(f"Tag not found: {org}/{repo}:{tag}") + + +class RepositoryApiError(RepositoryError): + """Temporary or unexpected API problem (network, timeout, 5xx, bad response). + + Nothing is wrong with the repository itself, so retry on the next run. + """ + + def __init__(self, service: str, detail: str): + super().__init__(f"{service} API error: {detail}") + + +class RateLimitedError(RepositoryApiError): + """The registry asked us to slow down. Stop the run instead of continuing.""" + status_code = 503 + code = "drq_rate_limited" + def __init__(self, service: str, retry_after: int | None = None): + self.retry_after = retry_after + hint = f", retry after {retry_after}s" if retry_after is not None else "" + super().__init__(service, f"rate limited{hint}") + +class OrganisationNotFoundError(RepositoryError): + """The organisation (namespace) has no repositories visible to us, so it most likely does not exist.""" + status_code = 422 + code = "drq_organisation_not_found" + client_visible = True + def __init__(self, org: str): + self.org = org + super().__init__(f"Organisation not found: {org}") \ No newline at end of file -- 2.43.0 From e06a9f522405bc58d4e808b9ca29c822a4699b4f Mon Sep 17 00:00:00 2001 From: Florian Date: Tue, 3 Nov 2026 10:41:14 +0100 Subject: [PATCH 2/2] - Added general authentication that seperates user calles from a gateway and cron scheduler updates - Added notifications being seperate by user_id - Added user paths to subscribe/unsubscribe from a repository - Cleaned up main.py by moving handlers to its own module --- database.sql | 28 ++++ src/handlers.py | 66 ++++++++++ src/main.py | 302 +++++++++++++++++++++++++++++++------------- src/models.py | 52 ++++++++ src/repositories.py | 123 ++++++++++++++++++ 5 files changed, 484 insertions(+), 87 deletions(-) create mode 100644 database.sql create mode 100644 src/handlers.py create mode 100644 src/models.py create mode 100644 src/repositories.py diff --git a/database.sql b/database.sql new file mode 100644 index 0000000..af076d4 --- /dev/null +++ b/database.sql @@ -0,0 +1,28 @@ +CREATE TABLE `watched_repositories` ( + `id` int unsigned NOT NULL AUTO_INCREMENT, + `type` enum('github','dockerhub') COLLATE utf8mb4_unicode_ci NOT NULL, + `org` varchar(39) COLLATE utf8mb4_unicode_ci NOT NULL, + `repo` varchar(100) COLLATE utf8mb4_unicode_ci NOT NULL, + `tag` varchar(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL, + `latest_entry` varchar(255) COLLATE utf8mb4_unicode_ci DEFAULT NULL, + `active` tinyint(1) NOT NULL DEFAULT '1', + `missing_checks` tinyint unsigned NOT NULL DEFAULT '0', + `created_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + UNIQUE KEY `uq_repository` (`type`,`org`,`repo`,`tag`), + KEY `idx_active_type` (`active`,`type`) +) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE `watched_repositories_subscriptions` ( + `id` int unsigned NOT NULL AUTO_INCREMENT, + `user_id` bigint NOT NULL, + `repository_id` int unsigned NOT NULL, + `custom_name` varchar(255) COLLATE utf8mb4_unicode_ci DEFAULT NULL, + `active` tinyint(1) NOT NULL DEFAULT '1', + `created_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + UNIQUE KEY `uq_user_repository` (`user_id`,`repository_id`), + KEY `idx_repository_active` (`repository_id`,`active`), + CONSTRAINT `fk_wrs_repository` FOREIGN KEY (`repository_id`) REFERENCES `watched_repositories` (`id`) ON DELETE CASCADE, + CONSTRAINT `fk_wrs_user` FOREIGN KEY (`user_id`) REFERENCES `users` (`user_id`) ON DELETE CASCADE +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci ; \ No newline at end of file diff --git a/src/handlers.py b/src/handlers.py new file mode 100644 index 0000000..8f12e2e --- /dev/null +++ b/src/handlers.py @@ -0,0 +1,66 @@ +from fastapi import FastAPI, Request +from fastapi.exceptions import RequestValidationError +from fastapi.responses import JSONResponse +from starlette.middleware.base import BaseHTTPMiddleware + +from metrics_server import REQUEST_COUNTER +from repository_errors import RepositoryError +from simple_logger_handler import setup_logger + +logger = setup_logger(__name__) + + +async def repository_error_handler(request: Request, exc: RepositoryError): + log = logger.warning if exc.client_visible else logger.error + log(f"[FastAPI] {request.method} {request.url.path} failed: {exc}") + return JSONResponse( + status_code=exc.status_code, + content={"code": exc.code, "detail": exc.client_message}, + ) + + +async def validation_error_handler(request: Request, exc: RequestValidationError): + errors = exc.errors() + + for err in errors: + original = (err.get("ctx") or {}).get("error") + if isinstance(original, RepositoryError): + return JSONResponse( + status_code=original.status_code, + content={"code": original.code, "detail": original.client_message}, + ) + + # Problems in headers, path or query are the calling service's fault, not the user's + if any(err["loc"][0] != "body" for err in errors): + logger.error(f"[FastAPI] Invalid request from a calling service: {[(e['loc'], e['msg']) for e in errors]}") + return JSONResponse(status_code=400, content={"code": "bad_request", "detail": "The request could not be processed"}) + + messages = [] + for err in errors: + field = ".".join(str(p) for p in err["loc"] if p != "body") + msg = err["msg"].removeprefix("Value error, ") + messages.append(f"{field}: {msg}" if field else msg) + return JSONResponse(status_code=422, content={"code": "validation_error", "detail": "; ".join(messages)}) + + +async def prometheus_middleware(request: Request, call_next): + logger.debug(f"[FastAPI] Incoming request: {request.method} {request.url.path}") + status = 500 + try: + response = await call_next(request) + status = response.status_code + logger.debug(f"[FastAPI] Request processed with status {status}") + except Exception as e: + logger.error(f"[FastAPI] Exception occurred: {e}", exc_info=True) + raise + finally: + REQUEST_COUNTER.labels(request.method, request.url.path, status).inc() + logger.debug(f"[FastAPI] Counter incremented for {request.method} {request.url.path} [{status}]") + return response + + +def register_handlers(app: FastAPI) -> None: + """Attach the error handlers and the metrics middleware to the app.""" + app.add_exception_handler(RepositoryError, repository_error_handler) + app.add_exception_handler(RequestValidationError, validation_error_handler) + app.add_middleware(BaseHTTPMiddleware, dispatch=prometheus_middleware) diff --git a/src/main.py b/src/main.py index d00298b..4c71a9d 100644 --- a/src/main.py +++ b/src/main.py @@ -3,20 +3,33 @@ import os from contextlib import asynccontextmanager import uvicorn -from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request, Response +from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request +from service_auth import UserId from auth_config import Caller, auth from db_module import close_connection_pool, create_connection_pool, get_db, start_healthcheck_thread -from dockerhub_api import find_package_version_with_tag as find_package_version_with_tag_dockerhub -from github_api import find_package_version_with_tag as find_package_version_with_tag_github -from metrics_server import REQUEST_COUNTER, metrics_api +from handlers import register_handlers +from metrics_server import metrics_api +from models import RepositoryType, RepositoryUpdate +from repositories import ( + fetch_latest, + reset_missing_checks, + unsubscribe_every_subscriber_from_repository, + validate_repository_combination_and_return_latest_tag, +) +from repository_errors import ( + ApiAuthenticationFailure, + RateLimitedError, + RepositoryApiError, + RepositoryNotFoundError, + RepositoryTagNotFoundError, +) from secret_manager import cleanup_secret_files -from send_notifications import NotificationClient +from send_notifications import NotificationClient, NotificationError, NotificationUnavailableError from simple_logger_handler import LOG_LEVEL, setup_logger logger = setup_logger(__name__) - @asynccontextmanager async def lifespan(app: FastAPI): logger.info("[FastAPI] Starting application...") @@ -51,103 +64,218 @@ api = FastAPI( dependencies=[Depends(auth.authenticate)], ) -@api.middleware("http") -async def prometheus_middleware(request: Request, call_next): - logger.debug(f"[FastAPI] Incoming request: {request.method} {request.url.path}") - status = 500 - try: - response = await call_next(request) - status = response.status_code - logger.debug(f"[FastAPI] Request processed with status {status}") - except Exception as e: - logger.error(f"[FastAPI] Exception occurred: {e}", exc_info=True) - raise - finally: - REQUEST_COUNTER.labels(request.method, request.url.path, status).inc() - logger.debug(f"[FastAPI] Counter incremented for {request.method} {request.url.path} [{status}]") - return response - +register_handlers(api) user_router = APIRouter(dependencies=[Depends(auth.require(Caller.GATEWAY))]) scheduler_router = APIRouter(dependencies=[Depends(auth.require(Caller.SCHEDULER))]) -@scheduler_router.post("/github") -def handle_suwayomi( - request: Request, - db = Depends(get_db) - ): + +@scheduler_router.post("/update/{repository_type}") +def update_watched_repositories( + request: Request, + repository_type: RepositoryType, + db=Depends(get_db), + ): + results = {"checked": 0, "new": 0, "errors": 0} + with db.cursor() as cursor: + cursor.execute( + "SELECT id, org, repo, tag, latest_entry, missing_checks " + "FROM watched_repositories WHERE type = %s AND active = 1", + (repository_type.value,), + ) + repositories = cursor.fetchall() + logger.debug(f"[FastAPI] Found {len(repositories)} active {repository_type.value} repositories to check.") + + for repository_id, org, repo, tag, latest_entry, missing_checks in repositories: + results["checked"] += 1 + try: + online_version = fetch_latest(repository_type, org, repo, tag) + except RateLimitedError as e: + logger.warning(f"[FastAPI] Rate limited, stopping the {repository_type.value} check: {e}") + results["errors"] += 1 + break + except ApiAuthenticationFailure as e: + logger.error(f"[FastAPI] {e}") + results["errors"] += 1 + break + except (RepositoryNotFoundError, RepositoryTagNotFoundError) as e: + logger.warning(f"[FastAPI] {e}, deactivating repository ID {repository_id}") + unsubscribe_every_subscriber_from_repository(db, request.app.state.notifier, repository_id, org, repo, tag) + continue + except RepositoryApiError as e: + logger.warning(f"[FastAPI] Lookup failed for ID {repository_id}: {e}") + results["errors"] += 1 + continue + except Exception: + logger.exception(f"[FastAPI] Unexpected error for ID {repository_id} ({org}/{repo}:{tag})") + results["errors"] += 1 + continue + + if missing_checks: + reset_missing_checks(db, repository_id) + if latest_entry == online_version: + continue + + try: + + logger.info(f"[FastAPI] New version detected for org={org}, repo={repo}, tag={tag}") + + with db.cursor() as cursor: + cursor.execute( + "SELECT user_id, custom_name " + "FROM watched_repositories_subscriptions " + "WHERE repository_id = %s AND active = 1", + (repository_id,), + ) + subscribers = cursor.fetchall() + + sent = 0 + failed = 0 + for user_id, custom_name in subscribers: + try: + request.app.state.notifier.send( + title=custom_name or f"{org}/{repo}", + body=f"New version {online_version} for tag {tag}", + category="utility", + user_id=user_id, + ) + sent += 1 + except NotificationUnavailableError: + failed = len(subscribers) - sent + logger.error( + f"[FastAPI] Notifications backend unavailable, skipping " + f"{failed} remaining user(s) for repo={org}/{repo}" + ) + break + except NotificationError as e: + failed += 1 + logger.error( + f"[FastAPI] Failed to send notification to user_id={user_id} for repo={org}/{repo} " + f"status={getattr(e, 'status_code', None)}" + ) + + if subscribers and sent == 0: + results["errors"] += 1 + logger.error(f"[FastAPI] All notifications failed for ID:{repository_id}, will retry later") + continue + + with db.cursor() as cursor: + cursor.execute( + "UPDATE watched_repositories SET latest_entry = %s WHERE id = %s", + (online_version, repository_id), + ) + db.commit() + logger.debug(f"[FastAPI] Notification sent and database updated for repository ID {repository_id}") + results["new"] += 1 + + except Exception as e: + db.rollback() + results["errors"] += 1 + logger.error( + f"[FastAPI] Failed for ID {repository_id} (org={org}, repo={repo}, tag={tag}): {e}", + exc_info=True, + ) + + if results["errors"]: + logger.warning(f"[FastAPI] {repository_type.value} check finished with errors: {results}") + raise HTTPException(status_code=500, detail=results) + logger.info(f"[FastAPI] {repository_type.value} check complete: {results}") + return results + +@user_router.post("/add/repository") +def add_repository_subscription( + user_id: UserId, + repository: RepositoryUpdate, + db=Depends(get_db), +): + name = f"{repository.org}/{repository.repo}:{repository.tag}" + logger.info(f"[FastAPI] Subscribe request: {repository.type.value} {name}, user_id={user_id}") + + with db.cursor() as cursor: + cursor.execute( + "SELECT id, active FROM watched_repositories " + "WHERE type = %s AND org = %s AND repo = %s AND tag = %s", + (repository.type.value, repository.org, repository.repo, repository.tag), + ) + known = cursor.fetchone() + + latest_entry = None + if known is None or not known[1]: + latest_entry = validate_repository_combination_and_return_latest_tag(repository) try: - logger.info("[FastAPI] Suwayomi handler invoked") - online_version = find_package_version_with_tag_github("Suwayomi", "tachidesk", "stable") - logger.debug(f"[FastAPI] Fetched latest Suwayomi version from GitHub: {online_version}") - with db.cursor() as cursor: - cursor.execute("SELECT latest_version FROM docker_repositories WHERE app='suwayomi'") - local_state = cursor.fetchone() - local_version = str(local_state[0]) if local_state and local_state[0] is not None else None - logger.debug(f"[FastAPI] Comparing versions: local='{local_version}' vs online='{online_version}'") - - if local_version != online_version: - logger.debug("[FastAPI] Version mismatch detected. Updating database.") - cursor.execute ("UPDATE docker_repositories SET latest_version=%s WHERE app='suwayomi'", - (online_version,)) - db.commit() - logger.info("[FastAPI] New Suwayomi version recorded in database") - request.app.state.notifier.send( - title="New Suwayomi version has been found", - body=f"Version {online_version}", - category="utility", - user_id=1, - ) - logger.debug("[FastAPI] Notification sent for Suwayomi update") - return Response(status_code=200) - - logger.debug("[FastAPI] No new Suwayomi version found") - return Response(status_code=204) + cursor.execute( + "INSERT INTO watched_repositories (type, org, repo, tag, latest_entry) " + "VALUES (%s, %s, %s, %s, %s) " + "ON DUPLICATE KEY UPDATE id = LAST_INSERT_ID(id), active = 1, missing_checks = 0, " + "latest_entry = COALESCE(VALUES(latest_entry), latest_entry)", + (repository.type.value, repository.org, repository.repo, repository.tag, latest_entry), + ) + repository_id = cursor.lastrowid + cursor.execute( + "INSERT INTO watched_repositories_subscriptions (user_id, repository_id, custom_name) " + "VALUES (%s, %s, %s) " + "ON DUPLICATE KEY UPDATE active = 1, custom_name = VALUES(custom_name)", + (user_id, repository_id, repository.custom_name), + ) + db.commit() except Exception as e: - logger.error(f"[FastAPI] Error in Suwayomi handler: {e}", exc_info=True) - raise HTTPException(status_code=500, detail=str(e)) from e + db.rollback() + logger.exception(f"[FastAPI] Failed to subscribe user_id={user_id} to {name}") + raise HTTPException(status_code=500) from e + logger.info(f"[FastAPI] user_id={user_id} subscribed to repository ID {repository_id} ({name})") + return {"status": "ok", "detail": "Subscribed to repository", "name": name} +@user_router.post("/delete/repository") +def delete_repository_subscription( + user_id: UserId, + repository: RepositoryUpdate, + db=Depends(get_db), +): + name = f"{repository.org}/{repository.repo}:{repository.tag}" + logger.info(f"[FastAPI] Unsubscribe request: {repository.type.value} {name}, user_id={user_id}") -@scheduler_router.post("/dockerhub") -def handle_pihole( - request: Request, - db = Depends(get_db) - ): try: - logger.info("[FastAPI] Pi-hole handler invoked") - online_version = find_package_version_with_tag_dockerhub("pihole/pihole", "latest") - logger.debug(f"[FastAPI] Fetched latest Pi-hole version from Docker Hub: {online_version}") - with db.cursor() as cursor: - cursor.execute("SELECT latest_version FROM docker_repositories WHERE app='pihole'") - local_state = cursor.fetchone() - local_version = str(local_state[0]) if local_state and local_state[0] is not None else None - logger.debug(f"[FastAPI] Comparing versions: local='{local_version}' vs online='{online_version}'") + cursor.execute( + "SELECT id FROM watched_repositories " + "WHERE type = %s AND org = %s AND repo = %s AND tag = %s FOR UPDATE", + (repository.type.value, repository.org, repository.repo, repository.tag), + ) + known = cursor.fetchone() - if local_version != online_version: - logger.debug("[FastAPI] Version mismatch detected. Updating database.") - cursor.execute ("UPDATE docker_repositories SET latest_version=%s WHERE app='pihole'", - (online_version,)) - db.commit() - logger.info("[FastAPI] New Pi-hole version recorded in database") - request.app.state.notifier.send( - title="New Pi-hole version has been found", - body=f"Version {online_version}", - category="utility", - user_id=1, + if known is None: + db.rollback() + logger.info(f"[FastAPI] Repository {repository.type.value} {name} is not tracked, nothing to do") + return {"status":"ok","detail":"Unsubscribed from repository","name":name} + + repository_id = known[0] + cursor.execute( + "UPDATE watched_repositories_subscriptions SET active = 0 " + "WHERE user_id = %s AND repository_id = %s", + (user_id, repository_id), + ) + + cursor.execute( + "SELECT 1 from watched_repositories_subscriptions " + "WHERE repository_id = %s AND active = 1 LIMIT 1 FOR SHARE", + (repository_id,), + ) + if cursor.fetchone() is None: + cursor.execute( + "UPDATE watched_repositories SET active = 0 WHERE id = %s", + (repository_id,), ) - logger.debug("[FastAPI] Notification sent for Pi-hole update") - return Response(status_code=200) - - logger.debug("[FastAPI] No new Pi-hole version found") - return Response(status_code=204) - + db.commit() except Exception as e: - logger.error(f"[FastAPI] Error in Pi-hole handler: {e}", exc_info=True) - raise HTTPException(status_code=500, detail=str(e)) from e + db.rollback() + logger.exception(f"[FastAPI] Failed to unsubscribe user_id={user_id} to {name}") + raise HTTPException(status_code=500) from e + + logger.info(f"[FastAPI] Unsubscribe request successful: {repository.type.value} {name}, user_id={user_id}") + return {"status":"ok","detail":"Unsubscribed from repository","name":name} api.include_router(user_router) api.include_router(scheduler_router) diff --git a/src/models.py b/src/models.py new file mode 100644 index 0000000..a27002b --- /dev/null +++ b/src/models.py @@ -0,0 +1,52 @@ +import string +from enum import StrEnum + +from pydantic import BaseModel, Field, field_validator + +from repository_errors import InvalidOrganisationError, InvalidRepositoryError, InvalidTagError + +_REPO_CHARS = frozenset(string.ascii_letters + string.digits + "_.-") +_MAX_ORG = 39 +_MAX_REPO = 255 +_MAX_TAG = 128 + +class RepositoryType(StrEnum): + GITHUB = "github" + DOCKERHUB = "dockerhub" + +class RepositoryUpdate(BaseModel): + type: RepositoryType + org: str = Field(max_length=_MAX_ORG) + repo: str = Field(max_length=_MAX_REPO) + tag: str = Field(max_length=_MAX_TAG) + custom_name: str | None = Field(default=None, max_length=255) + + @field_validator("org") + @classmethod + def _validate_org(cls, v: str) -> str: + v = v.strip() + if v and _REPO_CHARS.issuperset(v) and v[0] not in ".-": + return v + raise InvalidOrganisationError(v) + + @field_validator("repo") + @classmethod + def _validate_repo(cls, v: str) -> str: + v = v.strip() + if v and _REPO_CHARS.issuperset(v) and v[0] not in ".-": + return v + raise InvalidRepositoryError(v) + + @field_validator("tag") + @classmethod + def _validate_tag(cls, v: str) -> str: + v = v.strip() + if v and _REPO_CHARS.issuperset(v) and v[0] not in ".-": + return v + raise InvalidTagError(v) + + @field_validator("custom_name") + @classmethod + def _blank_name_is_none(cls, v: str | None) -> str | None: + return (v or "").strip() or None + diff --git a/src/repositories.py b/src/repositories.py new file mode 100644 index 0000000..a0c798b --- /dev/null +++ b/src/repositories.py @@ -0,0 +1,123 @@ +from typing import assert_never + +from dockerhub_api import ( + find_package_version_with_tag as find_package_version_with_tag_dockerhub, +) +from dockerhub_api import ( + validate_repository_combination_and_return_latest_tag as validate_repo_dockerhub, +) +from github_api import ( + find_package_version_with_tag as find_package_version_with_tag_github, +) +from github_api import ( + validate_repository_combination_and_return_latest_tag as validate_repo_github, +) +from models import RepositoryType, RepositoryUpdate +from send_notifications import NotificationError, NotificationUnavailableError +from simple_logger_handler import setup_logger + +logger = setup_logger(__name__) + +_MAX_MISSING_CHECKS = 4 + +def fetch_latest(repository_type: RepositoryType, org:str, repo:str, tag:str) -> str: + match repository_type: + case RepositoryType.GITHUB: + return find_package_version_with_tag_github(org, repo, tag) + case RepositoryType.DOCKERHUB: + return find_package_version_with_tag_dockerhub(org, repo, tag) + case _: + assert_never(repository_type) + +def validate_repository_combination_and_return_latest_tag(repository: RepositoryUpdate) -> str: + match repository.type: + case RepositoryType.DOCKERHUB: + return validate_repo_dockerhub(repository.org, repository.repo, repository.tag) + case RepositoryType.GITHUB: + return validate_repo_github(repository.org, repository.repo, repository.tag) + case _: + assert_never(repository.type) + +def unsubscribe_every_subscriber_from_repository( + db, notifier, repository_id: int, org: str, repo: str, tag: str +) -> None: + """Count a "not found". On the 4th miss in a row, deactivate the repository and tell its subscribers.""" + subscribers = [] + try: + with db.cursor() as cursor: + cursor.execute( + "SELECT missing_checks FROM watched_repositories WHERE id = %s FOR UPDATE", + (repository_id,), + ) + row = cursor.fetchone() + if row is None: + db.rollback() + return + misses = row[0] + 1 + + if misses < _MAX_MISSING_CHECKS: + cursor.execute( + "UPDATE watched_repositories SET missing_checks = %s WHERE id = %s", + (misses, repository_id), + ) + else: + cursor.execute( + "SELECT user_id, custom_name FROM watched_repositories_subscriptions " + "WHERE repository_id = %s AND active = 1", + (repository_id,), + ) + subscribers = cursor.fetchall() + cursor.execute( + "UPDATE watched_repositories_subscriptions SET active = 0 WHERE repository_id = %s", + (repository_id,), + ) + cursor.execute( + "UPDATE watched_repositories SET active = 0, missing_checks = 0 WHERE id = %s", + (repository_id,), + ) + db.commit() + except Exception: + db.rollback() + logger.exception(f"[FastAPI] Failed to record a missing check for repository ID {repository_id}") + return + + if misses < _MAX_MISSING_CHECKS: + logger.warning( + f"[FastAPI] {org}/{repo}:{tag} not found ({misses}/{_MAX_MISSING_CHECKS}), " + f"checking again on the next run" + ) + return + + logger.warning( + f"[FastAPI] {org}/{repo}:{tag} not found {_MAX_MISSING_CHECKS} times in a row, deactivated " + f"repository ID {repository_id} and {len(subscribers)} subscription(s)" + ) + + for user_id, custom_name in subscribers: + try: + notifier.send( + title=custom_name or f"{org}/{repo}", + body=f"{org}/{repo}:{tag} no longer exists, so you were unsubscribed from it", + category="utility", + user_id=user_id, + ) + except NotificationUnavailableError: + logger.warning("[FastAPI] Notifications backend unavailable, not informing the remaining users") + break + except NotificationError as e: + logger.warning( + f"[FastAPI] Could not inform user_id={user_id} about the removal " + f"status={getattr(e, 'status_code', None)}" + ) + +def reset_missing_checks(db, repository_id: int) -> None: + try: + with db.cursor() as cursor: + cursor.execute( + "UPDATE watched_repositories SET missing_checks = 0 WHERE id = %s", + (repository_id,), + ) + db.commit() + except Exception: + db.rollback() + logger.exception(f"[FastAPI] Failed to reset the missing counter for repository ID {repository_id}") -- 2.43.0