"""Network fetches, with an on-disk cache.
Two things the front end wants that are not on the machine: an advisory's full
record from the security tracker, and the current PKGBUILD of a package paru has
never downloaded.
Both are cached under `~/.cache/updater`. Advisory records change rarely and the
tracker is a small volunteer service, so hammering it on every keypress would be
rude; the PKGBUILD cache is short-lived because it is expected to change.
Nothing here raises. A fetch that fails is a normal state for a tool that has to
work on a train, so every function returns a reason instead.
"""
from __future__ import annotations
import json
import os
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path
TRACKER = "https://security.archlinux.org"
AUR = "https://aur.archlinux.org"
#: advisories are effectively immutable once published
ADVISORY_TTL = 7 * 24 * 3600
#: a PKGBUILD can change at any moment, so keep it only briefly
PKGBUILD_TTL = 15 * 60
USER_AGENT = "updater/0.1 (+https://aur.archlinux.org)"
def cache_dir() -> Path:
base = os.environ.get("XDG_CACHE_HOME") or (Path.home() / ".cache")
return Path(base) / "updater"
def _cached(path: Path, ttl: int) -> str | None:
try:
if time.time() - path.stat().st_mtime > ttl:
return None
return path.read_text()
except OSError:
return None
def _store(path: Path, body: str) -> None:
try:
path.parent.mkdir(parents=True, exist_ok=True)
tmp = path.with_suffix(".tmp")
tmp.write_text(body)
tmp.replace(path)
except OSError:
pass # a cache that cannot be written is not a failure
def get(
url: str, *, timeout: float = 20, cache: Path | None = None, ttl: int = 0
) -> tuple[str, str]:
"""Fetch a URL. Returns (body, error). Exactly one of the two is empty."""
if cache is not None and ttl:
hit = _cached(cache, ttl)
if hit is not None:
return hit, ""
request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
try:
with urllib.request.urlopen(request, timeout=timeout) as response: # noqa: S310
body = response.read().decode("utf-8", errors="replace")
except urllib.error.HTTPError as exc:
return "", f"{url} returned HTTP {exc.code}"
except urllib.error.URLError as exc:
return "", f"could not reach {urllib.parse.urlsplit(url).netloc}: {exc.reason}"
except (TimeoutError, OSError) as exc:
return "", f"could not fetch {url}: {exc}"
if cache is not None:
_store(cache, body)
return body, ""
# --------------------------------------------------------------------------
# security tracker
# --------------------------------------------------------------------------
def advisory_detail(advisory_id: str, *, timeout: float = 20) -> dict:
"""The tracker's own record for one advisory.
`AVG-1234.json` exists, so this is structured data rather than a page to
scrape. It carries `affected` and `fixed` versions, the CVE list, any
references the tracker recorded, and its notes.
"""
if not advisory_id or "/" in advisory_id:
return {"id": advisory_id, "available": False, "reason": "not a usable advisory id"}
path = cache_dir() / "advisories" / f"{advisory_id}.json"
body, error = get(
f"{TRACKER}/{urllib.parse.quote(advisory_id)}.json",
timeout=timeout,
cache=path,
ttl=ADVISORY_TTL,
)
if error:
# fall back to a stale copy rather than showing nothing
if path.is_file():
try:
body = path.read_text()
except OSError:
return {"id": advisory_id, "available": False, "reason": error}
else:
return {"id": advisory_id, "available": False, "reason": error}
try:
raw = json.loads(body)
except ValueError as exc:
return {
"id": advisory_id,
"available": False,
"reason": f"the tracker returned something unreadable: {exc}",
}
if not isinstance(raw, dict):
return {"id": advisory_id, "available": False, "reason": "unexpected advisory shape"}
return {
"id": raw.get("name") or advisory_id,
"available": True,
"reason": "",
"packages": raw.get("packages") or [],
"status": raw.get("status") or "",
"severity": raw.get("severity") or "Unknown",
"kind": raw.get("type") or "",
"affected": raw.get("affected") or "",
"fixed": raw.get("fixed") or "",
"ticket": raw.get("ticket") or "",
"issues": raw.get("issues") or [],
"advisories": raw.get("advisories") or [],
"references": raw.get("references") or [],
"notes": raw.get("notes") or "",
"url": f"{TRACKER}/{raw.get('name') or advisory_id}",
}
# --------------------------------------------------------------------------
# AUR
# --------------------------------------------------------------------------
def pkgbuild_file(package: str, filename: str, *, timeout: float = 20.0) -> tuple[str, str]:
"""One file from an AUR repository, e.g. a `.install` scriptlet.
cgit serves any tracked path, so the same endpoint that returns the
PKGBUILD returns everything beside it.
"""
# Both parts come from the AUR, and the filename comes from the PKGBUILD's
# own `install=` value. Joined raw, an absolute path resets the join and
# `..` walks out of the cache directory -- and the fetched body is then
# written to whatever path that produced.
name = Path(filename).name
base = Path(package).name
if not name or not base or name in {".", ".."} or base in {".", ".."}:
return "", "the AUR repository named a file this cannot safely cache"
url = (
f"https://aur.archlinux.org/cgit/aur.git/plain/"
f"{urllib.parse.quote(name)}?h={urllib.parse.quote(base)}"
)
cache = cache_dir() / "aur" / base / name
body, error = get(url, timeout=timeout, cache=cache, ttl=PKGBUILD_TTL)
return (body or "", error or "")
def pkgbuild(package: str, *, timeout: float = 20) -> tuple[str, str]:
"""The current PKGBUILD for a package, straight from the AUR.
Used when paru has no clone to read: there is no history to diff against,
but showing the file that will actually be built is still worth doing.
"""
safe = urllib.parse.quote(package, safe="")
path = cache_dir() / "pkgbuilds" / f"{safe}.txt"
return get(
f"{AUR}/cgit/aur.git/plain/PKGBUILD?h={safe}", timeout=timeout, cache=path, ttl=PKGBUILD_TTL
)