#!/usr/bin/env python3 """Precyzyjny audyt http:// w polach source/sources/url receptur. Dla kazdego wystapienia: 1. rozwija placeholdery ${pkgname}/${pkgver}/... z receptury, 2. testuje https:// (kandydat) ORAZ http:// (oryginal), 3. klasyfikuje: SAFE_HTTPS - https dziala -> mozna podmienic HTTP_ONLY - http dziala, https nie -> NIE podmieniac BOTH_DEAD - oba nie dzialaja -> zrodlo martwe (osobny raport) UNRESOLVED - nie udalo sie rozwinac placeholderow Wyniki zapisywane przyrostowo do /tmp/http-urlaudit.json. """ import collections import concurrent.futures as cf import json import os import re import ssl import sys import threading import time import urllib.error import urllib.request import yaml R = "/var/lib/pagan-sync/recipes" OUT = "/tmp/http-urlaudit.json" CACHE = "/tmp/http-probe-cache.json" KEYS = ("source", "sources", "url") PH = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}") UA = {"User-Agent": "curl/8.7.1"} _lock = threading.Lock() cache = {} cache_dirty = 0 def as_list(v): if v is None: return [] return v if isinstance(v, list) else [v] def expand(s, d): """Rozwija ${...} z receptury. Zwraca (wynik, czy_zostaly_placeholdery).""" missing = [] def rep(m): k = m.group(1) if k in d and not isinstance(d[k], (dict, list)): return str(d[k]) missing.append(k) return m.group(0) out = PH.sub(rep, s) return out, bool(missing) def probe(url, timeout=20): """(ok, status). HEAD -> GET z Range (nie sciaga pliku).""" ctx = ssl.create_default_context() last = None def attempt(method, extra=None): h = dict(UA) if extra: h.update(extra) req = urllib.request.Request(url, headers=h, method=method) with urllib.request.urlopen(req, timeout=timeout, context=ctx) as r: return r.status try: st = attempt("HEAD") if 200 <= st < 400: return True, st last = f"HTTP {st}" except urllib.error.HTTPError as e: last = f"HTTP {e.code}" if e.code not in (403, 405, 400, 501): return False, last except Exception as e: last = type(e).__name__ try: st = attempt("GET", {"Range": "bytes=0-1"}) if 200 <= st < 400: return True, st return False, f"HTTP {st}" except urllib.error.HTTPError as e: if e.code in (403, 405): # serwer istnieje, tylko nie lubi naszych metod return True, f"HTTP {e.code}" return False, f"HTTP {e.code}" except Exception as e: return False, f"{last or ''}/{type(e).__name__}".strip("/") def cached_probe(url): with _lock: if url in cache: return tuple(cache[url]) res = probe(url) with _lock: cache[url] = list(res) return res def save_cache(): with _lock: snap = dict(cache) tmp = CACHE + ".tmp" json.dump(snap, open(tmp, "w")) os.replace(tmp, CACHE) def main(): global cache if os.path.exists(CACHE): try: cache = json.load(open(CACHE)) except Exception: cache = {} occ = [] # (rel, key, raw, expanded, unresolved) for dp, _d, fs in os.walk(R): if "PAGBUILD.yaml" not in fs: continue p = os.path.join(dp, "PAGBUILD.yaml") rel = p[len(R) + 1:] try: d = yaml.safe_load(open(p, encoding="utf-8")) or {} except Exception: continue for k in KEYS: for v in as_list(d.get(k)): s = str(v) if s.startswith("http://"): ex, unres = expand(s, d) occ.append((rel, k, s, ex, unres)) uniq = sorted({o[3] for o in occ if not o[4]}) print(f"wystapien http://: {len(occ)} | unikalnych rozwinietch URL: {len(uniq)}", flush=True) results = {} t0 = time.time() done = 0 to_probe = [] for ex in uniq: if ex in cache: continue to_probe.append(ex) with cf.ThreadPoolExecutor(max_workers=12) as pool: futs = {pool.submit(cached_probe, u): u for u in to_probe} for fut in cf.as_completed(futs): u = futs[fut] try: res = fut.result() except Exception as e: res = (False, type(e).__name__) results[u] = list(res) done += 1 if done % 50 == 0: print(f" ... {done}/{len(to_probe)} ({time.time()-t0:.0f}s)", flush=True) save_cache() # dociagnij z cache te, ktore juz byly for u in uniq: if u not in results and u in cache: results[u] = list(cache[u]) save_cache() # dla kazdego URL sprawdz tez oryginalne http http_urls = sorted({o[3] for o in occ if not o[4]}) http_res = {} to_p = [u for u in http_urls if u not in cache] print(f"testowanie oryginalow http:// (bez zmian): {len(to_p)}", flush=True) with cf.ThreadPoolExecutor(max_workers=12) as pool: futs = {pool.submit(cached_probe, u): u for u in to_p} for fut in cf.as_completed(futs): u = futs[fut] try: http_res[u] = list(fut.result()) except Exception as e: http_res[u] = [False, type(e).__name__] for u in http_urls: if u not in http_res: http_res[u] = list(cache.get(u, [False, "?"])) save_cache() out = collections.defaultdict(list) for rel, k, raw, ex, unres in occ: if unres: cls = "UNRESOLVED" hs = http_res.get(ex, [False, "?"])[1] else: hs = ex # placeholder: klucz https_ok = results.get(ex, [False, "?"])[0] http_ok = http_res.get(ex, [False, "?"])[0] if https_ok: cls = "SAFE_HTTPS" elif http_ok: cls = "HTTP_ONLY" else: cls = "BOTH_DEAD" hs = f"https={results.get(ex, [False,'?'])[1]} http={http_res.get(ex, [False,'?'])[1]}" out[cls].append({"recipe": rel, "key": k, "raw": raw, "expanded": ex, "detail": hs}) json.dump(out, open(OUT, "w"), indent=1, ensure_ascii=False) print() for cls in ("SAFE_HTTPS", "HTTP_ONLY", "BOTH_DEAD", "UNRESOLVED"): print(f"{cls:12s}: {len(out.get(cls, []))} wystapien") hosts = collections.Counter() for item in out.get("HTTP_ONLY", []): m = re.match(r"^https?://([^/]+)", item["expanded"]) if m: hosts[m.group(1)] += 1 if hosts: print("\nHTTP_ONLY (NIE podmieniac) wg hosta:") for h, n in hosts.most_common(): print(f" {n:4d} {h}") hosts = collections.Counter() for item in out.get("SAFE_HTTPS", []): m = re.match(r"^https?://([^/]+)", item["expanded"]) if m: hosts[m.group(1)] += 1 if hosts: print("\nSAFE_HTTPS (do podmiany) wg hosta:") for h, n in hosts.most_common(25): print(f" {n:4d} {h}") print(f"\nzapisano {OUT} (szczegoly) i {CACHE} (cache sond)") if __name__ == "__main__": sys.exit(main())