90 lines
3.5 KiB
Python
90 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
"""rt_candidates.py OUT.json APP=IMAGE:TAG ... — for each pin with a bad runtime: the registry's tags in the same
|
|
shape as the pinned tag, the newest in the same main version and the newest overall, and the RUNTIME read from each of
|
|
those images (image_runtime.read_image — the image, never release notes), judged with the watch's own table."""
|
|
import json, os, re, sys, urllib.request, urllib.error
|
|
|
|
sys.path.insert(0, "/mnt/5_hdd/felhom.eu/git/felhom.eu/scripts")
|
|
import image_runtime as ir # noqa: E402
|
|
import security_watch as sw # noqa: E402
|
|
|
|
|
|
def tags(ref):
|
|
reg, path, _ = ir.split_ref(ref)
|
|
r = ir.Registry(reg, path)
|
|
url, out = "%s/v2/%s/tags/list?n=1000" % (r.base, path), []
|
|
while url:
|
|
req = urllib.request.Request(url, headers=dict({"User-Agent": ir.UA}, **r.auth))
|
|
resp = urllib.request.urlopen(req, timeout=60)
|
|
out += json.loads(resp.read()).get("tags") or []
|
|
link = resp.headers.get("Link", "")
|
|
m = re.search(r"<([^>]+)>;\s*rel=\"next\"", link)
|
|
url = (m.group(1) if m.group(1).startswith("http") else r.base + m.group(1)) if m else None
|
|
return out
|
|
|
|
|
|
def shape(tag):
|
|
return re.compile("^" + re.sub(r"\d+", r"(\\d+)", re.escape(tag)) + "$")
|
|
|
|
|
|
def nums(tag, rx):
|
|
m = rx.match(tag)
|
|
return tuple(int(x) for x in m.groups()) if m else None
|
|
|
|
|
|
def verdict(res):
|
|
if res.get("unchecked"):
|
|
return "NOT CHECKED: " + "; ".join(res["unchecked"])[:120]
|
|
adv = sw.runtime_advisories(res) if hasattr(sw, "runtime_advisories") else None
|
|
return adv
|
|
|
|
|
|
def main():
|
|
import concurrent.futures, threading
|
|
out, lock = {}, threading.Lock()
|
|
if os.path.isfile(sys.argv[1]):
|
|
out = json.load(open(sys.argv[1]))
|
|
todo = [a for a in sys.argv[2:] if a.split("=", 1)[0] not in out or "error" in out[a.split("=", 1)[0]]]
|
|
|
|
def one(arg):
|
|
rec = do(arg)
|
|
with lock:
|
|
out[arg.split("=", 1)[0]] = rec
|
|
json.dump(out, open(sys.argv[1], "w"), indent=1)
|
|
with concurrent.futures.ThreadPoolExecutor(5) as ex:
|
|
list(ex.map(one, todo))
|
|
return 0
|
|
|
|
|
|
def do(arg):
|
|
if True:
|
|
app, ref = arg.split("=", 1)
|
|
img, tag = ref.rsplit(":", 1)
|
|
rx = shape(tag)
|
|
cur = nums(tag, rx)
|
|
try:
|
|
ts = [t for t in tags(ref) if nums(t, rx)]
|
|
except Exception as e: # noqa: BLE001
|
|
print(app, "TAGS ERROR", e, flush=True)
|
|
return {"ref": ref, "error": "tags: %s" % e}
|
|
newer = sorted((t for t in ts if nums(t, rx) > cur), key=lambda t: nums(t, rx))
|
|
same = [t for t in newer if nums(t, rx)[0] == cur[0]]
|
|
pick = []
|
|
if same:
|
|
pick.append(same[-1])
|
|
if newer and newer[-1] not in pick:
|
|
pick.append(newer[-1])
|
|
rec = {"ref": ref, "newer": newer[-12:], "newest_same_main": same[-1] if same else None,
|
|
"newest": newer[-1] if newer else None, "reads": {}}
|
|
for t in pick: # the pinned tag's runtime is known (the watch's cache, R-1332)
|
|
res = ir.read_image("%s:%s" % (img, t))
|
|
rec["reads"][t] = {k: res.get(k) for k in ("node", "brotli", "brotlicffi", "apk", "dpkg", "unchecked", "env")}
|
|
print(app, t, "node=%s brotli=%s cffi=%s apk=%s dpkg=%s %s" % (
|
|
res.get("node"), res.get("brotli"), res.get("brotlicffi"), res.get("apk"), res.get("dpkg"),
|
|
("UNCHECKED " + "; ".join(res["unchecked"])[:100]) if res.get("unchecked") else ""), flush=True)
|
|
return rec
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|