from __future__ import annotations import os import re import urllib.error import urllib.request from dataclasses import dataclass from typing import Any # Matches both the legacy Gitea-default form (.../raw//) and the # canonical branch-qualified form (.../raw/branch//) that both # Gitea and Forgejo serve without a 303 redirect. See REUSE-WP-0017-T06 for # why the redirect matters (urllib follows it fine; some HEAD-only probes # and shell tooling don't). _RAW_URL_RE = re.compile( r"^(?Phttps?)://(?P[^/]+)/(?P[^/]+)/(?P[^/]+)" r"/raw/(?:branch/)?(?P[^/]+)/(?P.+)$" ) @dataclass class RawUrlParts: scheme: str host: str org: str repo: str branch: str path: str def parse_raw_url(url: str) -> RawUrlParts: match = _RAW_URL_RE.match(url) if not match: raise ValueError(f"not a recognized Gitea/Forgejo raw URL: {url}") return RawUrlParts(**match.groupdict()) def derive_raw_url(base_url: str, org: str, repo: str, *, branch: str = "main", path: str = "registry/indexes/capabilities.yaml") -> str: """Builds a raw index URL in the canonical branch-qualified form, which both Gitea and Forgejo serve directly (no 303 redirect).""" base = base_url.rstrip("/") return f"{base}/{org}/{repo}/raw/branch/{branch}/{path}" def rewrite_url_host(url: str, new_base_url: str) -> str: """Rewrites a raw index URL's scheme+host to new_base_url, preserving org/repo/branch/path and normalizing to the canonical branch-qualified form regardless of which form the input used.""" parts = parse_raw_url(url) return derive_raw_url(new_base_url, parts.org, parts.repo, branch=parts.branch, path=parts.path) def forge_base_url(explicit: str | None = None) -> str | None: """Returns the configured default Forgejo base URL, or None if unset. Unlike hub_client.service_base_url, this has no hard requirement -- per-repo migration is opt-in (most repos haven't moved their remote yet), so an unset value just means 'no default target configured', not an error.""" return explicit or os.environ.get("REUSE_SURFACE_FORGE_BASE_URL") or None def probe_url(url: str, *, timeout: int = 10) -> tuple[bool, int | None]: """HEAD-probes a URL, following redirects (matches establish.py's _probe_raw_url behavior). Returns (ok, status).""" request = urllib.request.Request(url, method="HEAD", headers={"User-Agent": "reuse-surface/0.1"}) try: with urllib.request.urlopen(request, timeout=timeout) as response: return response.status == 200, response.status except urllib.error.HTTPError as exc: return False, exc.code except (urllib.error.URLError, TimeoutError, OSError): return False, None def migrate_source_host( sources: list[dict[str, Any]], repo: str, new_base_url: str, *, verify: bool = True, ) -> dict[str, Any]: """Rewrites one repo's entry in a sources.yaml `sources` list to point at new_base_url. Mutates the matching entry in place and returns a report dict. Raises ValueError if the repo isn't found, the existing URL can't be parsed, or (when verify=True) the new URL doesn't resolve -- never writes a URL that hasn't been confirmed reachable, since rewriting to a host the repo hasn't actually migrated to would silently break federation for it.""" entry = next((s for s in sources if s.get("repo") == repo), None) if entry is None: raise ValueError(f"repo not found in sources: {repo}") old_url = entry["url"] new_url = rewrite_url_host(old_url, new_base_url) report: dict[str, Any] = {"repo": repo, "old_url": old_url, "new_url": new_url, "verified": None} if verify: ok, status = probe_url(new_url) report["verified"] = ok report["probe_status"] = status if not ok: raise ValueError( f"{repo}: new URL did not resolve (HTTP {status}); " f"not writing an unverified URL: {new_url}" ) entry["url"] = new_url return report