lemmy/update.py

   1 #! /usr/bin/env nix-shell
   2 #! nix-shell -i python3 -p python3 python3.pkgs.semver nix-prefetch-github
   3 from urllib.request import Request, urlopen
   4 import dataclasses
   5 import subprocess
   6 import hashlib
   7 import os.path
   8 import semver
   9 import base64
  10 from typing import (
  11     Optional,
  12     Dict,
  13     List,
  14 )
  15 import json
  16 import os
  17
  18
  19 SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
  20 NIXPKGS = os.path.abspath(os.path.join(SCRIPT_DIR, "../../../../"))
  21
  22
  23 OWNER = "LemmyNet"
  24 UI_REPO = "lemmy-ui"
  25 SERVER_REPO = "lemmy"
  26
  27
  28 @dataclasses.dataclass
  29 class Pin:
  30     serverVersion: str
  31     uiVersion: str
  32     serverSha256: str = ""
  33     serverCargoSha256: str = ""
  34     uiSha256: str = ""
  35     uiYarnDepsSha256: str = ""
  36
  37     filename: Optional[str] = None
  38
  39     def write(self) -> None:
  40         if not self.filename:
  41             raise ValueError("No filename set")
  42
  43         with open(self.filename, "w") as fd:
  44             pin = dataclasses.asdict(self)
  45             del pin["filename"]
  46             json.dump(pin, fd, indent=2)
  47             fd.write("\n")
  48
  49
  50 def github_get(path: str) -> Dict:
  51     """Send a GET request to Gituhb, optionally adding GITHUB_TOKEN auth header"""
  52     url = f"https://api.github.com/{path.lstrip('/')}"
  53     print(f"Retreiving {url}")
  54
  55     req = Request(url)
  56
  57     if "GITHUB_TOKEN" in os.environ:
  58         req.add_header("authorization", f"Bearer {os.environ['GITHUB_TOKEN']}")
  59
  60     with urlopen(req) as resp:
  61         return json.loads(resp.read())
  62
  63
  64 def get_latest_release(owner: str, repo: str) -> str:
  65     return github_get(f"/repos/{owner}/{repo}/releases/latest")["tag_name"]
  66
  67
  68 def sha256_url(url: str) -> str:
  69     sha256 = hashlib.sha256()
  70     with urlopen(url) as resp:
  71         while data := resp.read(1024):
  72             sha256.update(data)
  73     return "sha256-" + base64.urlsafe_b64encode(sha256.digest()).decode()
  74
  75
  76 def prefetch_github(owner: str, repo: str, rev: str) -> str:
  77     """Prefetch github rev and return sha256 hash"""
  78     print(f"Prefetching {owner}/{repo}({rev})")
  79
  80     proc = subprocess.run(
  81         ["nix-prefetch-github", owner, repo, "--rev", rev, "--fetch-submodules"],
  82         check=True,
  83         stdout=subprocess.PIPE,
  84     )
  85
  86     sha256 = json.loads(proc.stdout)["sha256"]
  87     if not sha256.startswith("sha256-"):  # Work around bug in nix-prefetch-github
  88         return "sha256-" + sha256
  89
  90     return sha256
  91
  92
  93 def get_latest_tag(owner: str, repo: str, prerelease: bool = False) -> str:
  94     """Get the latest tag from a Github Repo"""
  95     tags: List[str] = []
  96
  97     # As the Github API doesn't have any notion of "latest" for tags we need to
  98     # collect all of them and sort so we can figure out the latest one.
  99     i = 0
 100     while i <= 100:  # Prevent infinite looping
 101         i += 1
 102         resp = github_get(f"/repos/{owner}/{repo}/tags?page={i}")
 103         if not resp:
 104             break
 105
 106         # Filter out unparseable tags
 107         for tag in resp:
 108             try:
 109                 parsed = semver.Version.parse(tag["name"])
 110                 if (
 111                     semver.Version.parse(tag["name"])
 112                     and not prerelease
 113                     and parsed.prerelease
 114                 ):  # Filter out release candidates
 115                     continue
 116             except ValueError:
 117                 continue
 118             else:
 119                 tags.append(tag["name"])
 120
 121     # Sort and return latest
 122     return sorted(tags, key=lambda name: semver.Version.parse(name))[-1]
 123
 124
 125 def get_fod_hash(attr: str) -> str:
 126     """
 127     Get fixed output hash for attribute.
 128     This depends on a fixed output derivation with an empty hash.
 129     """
 130
 131     print(f"Getting fixed output hash for {attr}")
 132
 133     proc = subprocess.run(["nix-build", NIXPKGS, "-A", attr], stderr=subprocess.PIPE)
 134     if proc.returncode != 1:
 135         raise ValueError("Expected nix-build to fail")
 136
 137     # Iterate list in reverse order so we get the "got:" line early
 138     for line in proc.stderr.decode().split("\n")[::-1]:
 139         cols = line.split()
 140         if cols and cols[0] == "got:":
 141             return cols[1]
 142
 143     raise ValueError("No fixed output hash found")
 144
 145
 146 def make_server_pin(pin: Pin, attr: str) -> None:
 147     pin.serverSha256 = prefetch_github(OWNER, SERVER_REPO, pin.serverVersion)
 148     pin.write()
 149     pin.serverCargoSha256 = get_fod_hash(attr)
 150     pin.write()
 151
 152
 153 def make_ui_pin(pin: Pin, package_json: str, attr: str) -> None:
 154     # Save a copy of package.json
 155     print("Getting package.json")
 156     with urlopen(
 157         f"https://raw.githubusercontent.com/{OWNER}/{UI_REPO}/{pin.uiVersion}/package.json"
 158     ) as resp:
 159         with open(os.path.join(SCRIPT_DIR, package_json), "wb") as fd:
 160             fd.write(resp.read())
 161
 162     pin.uiSha256 = prefetch_github(OWNER, UI_REPO, pin.uiVersion)
 163     pin.write()
 164     pin.uiYarnDepsSha256 = get_fod_hash(attr)
 165     pin.write()
 166
 167
 168 if __name__ == "__main__":
 169     # Get server version
 170     server_version = get_latest_tag(OWNER, SERVER_REPO)
 171
 172     # Get UI version (not always the same as lemmy-server)
 173     ui_version = get_latest_tag(OWNER, UI_REPO)
 174
 175     pin = Pin(server_version, ui_version, filename=os.path.join(SCRIPT_DIR, "pin.json"))
 176     make_server_pin(pin, "lemmy-server")
 177     make_ui_pin(pin, "package.json", "lemmy-ui")