reuse-surface/reuse_surface/forge_host.py

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/<branch>/<path>) and the
# canonical branch-qualified form (.../raw/branch/<branch>/<path>) 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"^(?P<scheme>https?)://(?P<host>[^/]+)/(?P<org>[^/]+)/(?P<repo>[^/]+)"
    r"/raw/(?:branch/)?(?P<branch>[^/]+)/(?P<path>.+)$"
)


@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