oni.git

oni

~ A static Git browser written in Rust.
git clone https://github.com/debuman/oni.git

scripts/readme_assets.py (8044B)

      1 """Resolve repository-local links and images without changing README markup."""
      2 
      3 from html import escape
      4 from html.parser import HTMLParser
      5 import posixpath
      6 import subprocess
      7 from urllib.parse import quote, unquote, urlsplit, urlunsplit
      8 
      9 MAX_IMAGE_BYTES = 32 * 1024 * 1024
     10 
     11 
     12 def image_type(name, data):
     13     suffix = name.rsplit(".", 1)[-1].lower()
     14     valid = {
     15         "png": data.startswith(b"\x89PNG\r\n\x1a\n"),
     16         "jpg": data.startswith(b"\xff\xd8\xff"),
     17         "jpeg": data.startswith(b"\xff\xd8\xff"),
     18         "gif": data.startswith((b"GIF87a", b"GIF89a")),
     19         "webp": data.startswith(b"RIFF") and data[8:12] == b"WEBP",
     20         "ico": data.startswith(b"\x00\x00\x01\x00"),
     21         "avif": data[4:8] == b"ftyp" and data[8:12] in (b"avif", b"avis"),
     22     }
     23     return valid.get(suffix, False)
     24 
     25 
     26 def contained_path(root, name):
     27     if (name.startswith("/") or '\\' in name or ':' in name
     28             or any(part in ("", ".", "..") for part in name.split("/"))):
     29         raise ValueError("Unsafe repository asset path")
     30     root = root.resolve()
     31     current = root
     32     for part in name.split("/"):
     33         current = current / part
     34         if current.is_symlink() or current.is_junction():
     35             raise ValueError("Asset paths must not contain links")
     36     current.resolve().relative_to(root)
     37     return current
     38 
     39 
     40 class ReadmeLinks(HTMLParser):
     41     def __init__(self, output, git_dir, repo):
     42         super().__init__(convert_charrefs=False)
     43         self.output = output
     44         self.git_dir = git_dir
     45         self.repo = repo
     46         self.parts = []
     47         self.objects = None
     48         listing = subprocess.check_output(["git", "-C", str(git_dir), "ls-tree", "-rz", "HEAD"])
     49         self.blobs = {}
     50         self.directories = set()
     51         for entry in listing.split(b"\0"):
     52             if not entry:
     53                 continue
     54             metadata, name = entry.split(b"\t", 1)
     55             mode, kind, oid = metadata.decode().split()
     56             if kind == "blob" and mode in ("100644", "100755"):
     57                 name = name.decode("utf-8")
     58                 self.blobs[name] = oid
     59                 # Index ancestors once, so every README directory link is a
     60                 # set lookup instead of a scan of all repository files.
     61                 parent = posixpath.dirname(name)
     62                 while parent and parent not in self.directories:
     63                     self.directories.add(parent)
     64                     parent = posixpath.dirname(parent)
     65 
     66     def blob(self, oid):
     67         if self.objects is None:
     68             self.objects = subprocess.Popen(
     69                 ["git", "-C", str(self.git_dir), "cat-file", "--batch"],
     70                 stdin=subprocess.PIPE, stdout=subprocess.PIPE,
     71             )
     72         self.objects.stdin.write(oid.encode("ascii") + b"\n")
     73         self.objects.stdin.flush()
     74         header = self.objects.stdout.readline().split()
     75         if len(header) != 3 or header[0] != oid.encode("ascii") or header[1] != b"blob":
     76             raise RuntimeError("Invalid Git blob response")
     77         size = int(header[2])
     78         if size < 0 or size > MAX_IMAGE_BYTES:
     79             raise RuntimeError("Git image blob exceeds the 32 MiB limit")
     80         data = self.objects.stdout.read(size)
     81         if len(data) != size or self.objects.stdout.read(1) != b"\n":
     82             raise RuntimeError("Truncated Git blob response")
     83         return data
     84 
     85     def close(self):
     86         try:
     87             super().close()
     88         finally:
     89             if self.objects is not None:
     90                 objects, self.objects = self.objects, None
     91                 try:
     92                     objects.stdin.close()
     93                 except BrokenPipeError:
     94                     pass
     95                 finally:
     96                     objects.stdout.close()
     97                 if objects.wait() != 0:
     98                     raise RuntimeError("Git blob process failed")
     99 
    100     def resolve_link(self, value, is_image):
    101         try:
    102             url = urlsplit(value)
    103         except ValueError:
    104             return ""
    105         if url.scheme or url.netloc or not url.path or url.path.startswith("/"):
    106             return value
    107         name = posixpath.normpath(unquote(url.path))
    108         if name.startswith("../") or name == "..":
    109             return value
    110         try:
    111             existing = contained_path(self.output, name)
    112         except ValueError:
    113             return value
    114         if existing.is_file():
    115             return value
    116         if is_image and name in self.blobs:
    117             # SVG is an active document when opened directly. Keep it off this
    118             # origin; GitHub's raw host is separate from the generated site.
    119             if name.lower().endswith(".svg"):
    120                 oid = subprocess.check_output(["git", "-C", str(self.git_dir), "rev-parse", "HEAD"], timeout=30).decode().strip()
    121                 target = f"https://raw.githubusercontent.com/{self.repo['full_name']}/{oid}/{quote(name, safe='/')}"
    122                 result = urlsplit(target)
    123                 return urlunsplit((result.scheme, result.netloc, result.path, url.query, url.fragment))
    124             if name.rsplit(".", 1)[-1].lower() not in ("png", "jpg", "jpeg", "gif", "webp", "ico", "avif"):
    125                 return ""
    126             asset = contained_path(self.output, "raw/" + name)
    127             asset.parent.mkdir(parents=True, exist_ok=True)
    128             if not asset.exists():
    129                 data = self.blob(self.blobs[name])
    130                 if not image_type(name, data):
    131                     return ""
    132                 with asset.open("xb") as output:
    133                     output.write(data)
    134             target = "raw/" + quote(name, safe="/")
    135         elif name in self.blobs and (self.output / "file" / (name + ".html")).is_file():
    136             target = "file/" + quote(name, safe="/") + ".html"
    137         elif name.rstrip("/") in self.directories:
    138             target = (f"https://github.com/{self.repo['full_name']}/tree/"
    139                       f"{quote(self.repo['default_branch'], safe='')}/{quote(name, safe='/')}")
    140         else:
    141             return value
    142         result = urlsplit(target)
    143         return urlunsplit((result.scheme, result.netloc, result.path, url.query, url.fragment))
    144 
    145     def start_tag(self, tag, attrs, closed=False):
    146         updated = []
    147         changed = False
    148         for key, value in attrs:
    149             replacement = self.resolve_link(value, key == "src") if value and key in ("src", "href") else value
    150             changed |= replacement != value
    151             updated.append((key, replacement))
    152         if not changed:
    153             self.parts.append(self.get_starttag_text())
    154             return
    155         attributes = "".join(" " + key + (f'="{escape(value, quote=True)}"' if value is not None else "")
    156                              for key, value in updated)
    157         self.parts.append("<" + tag + attributes + (" />" if closed else ">"))
    158 
    159     def handle_starttag(self, tag, attrs):
    160         self.start_tag(tag, attrs)
    161 
    162     def handle_startendtag(self, tag, attrs):
    163         self.start_tag(tag, attrs, closed=True)
    164 
    165     def handle_endtag(self, tag):
    166         self.parts.append("</" + tag + ">")
    167 
    168     def handle_data(self, data):
    169         self.parts.append(data)
    170 
    171     def handle_entityref(self, name):
    172         self.parts.append("&" + name + ";")
    173 
    174     def handle_charref(self, name):
    175         self.parts.append("&#" + name + ";")
    176 
    177     def handle_comment(self, data):
    178         self.parts.append("<!--" + data + "-->")
    179 
    180     def handle_decl(self, decl):
    181         self.parts.append("<!" + decl + ">")
    182 
    183 
    184 def link_readme_assets(output, git_dir, repo):
    185     page = output / "README.html"
    186     if not page.exists():
    187         return
    188     document = page.read_text(encoding="utf-8")
    189     marker = '<article class="readme">\n'
    190     if marker not in document:
    191         return
    192     prefix, body = document.split(marker, 1)
    193     body, suffix = body.rsplit('</article>', 1)
    194     parser = ReadmeLinks(output, git_dir, repo)
    195     try:
    196         parser.feed(body)
    197     finally:
    198         parser.close()
    199     page.write_text(prefix + marker + "".join(parser.parts) + '</article>' + suffix, encoding="utf-8")