136 lines
4.8 KiB
Python
136 lines
4.8 KiB
Python
"""Verify public repository references without forwarding any credentials.
|
|
|
|
Only the school's Gitea and GitHub APIs are queried. Redirects, userinfo,
|
|
custom ports and unsupported git hosts never expand the network allowlist.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from functools import lru_cache
|
|
from html import unescape, escape
|
|
import json
|
|
import re
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
from tools.answers.context_policy import text_urls
|
|
|
|
|
|
class NoRedirect(urllib.request.HTTPRedirectHandler):
|
|
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
|
return None
|
|
|
|
|
|
def repository_reference(url):
|
|
try:
|
|
host = (urllib.parse.urlsplit(url).hostname or "").lower()
|
|
return host in {"github.com", "gitlab.com"} or host.startswith(
|
|
("git.", "gitlab.", "gitea.")
|
|
)
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
@lru_cache(maxsize=512)
|
|
def _probe(url, minute):
|
|
try:
|
|
parsed = urllib.parse.urlsplit(url)
|
|
if (
|
|
parsed.scheme != "https"
|
|
or parsed.username
|
|
or parsed.password
|
|
or parsed.port not in {None, 443}
|
|
or parsed.query
|
|
or parsed.hostname not in {"git.kemt.fei.tuke.sk", "github.com"}
|
|
):
|
|
return "unverified"
|
|
parts = parsed.path.strip("/").split("/")
|
|
if len(parts) < 2 or any(
|
|
not re.fullmatch(r"[A-Za-z0-9_.-]+", p) or p in {".", ".."} for p in parts[:2]
|
|
):
|
|
return "unverified"
|
|
owner, repo = parts[0], parts[1].removesuffix(".git")
|
|
base = (
|
|
"https://api.github.com/repos/"
|
|
if parsed.hostname == "github.com"
|
|
else "https://git.kemt.fei.tuke.sk/api/v1/repos/"
|
|
)
|
|
opener = urllib.request.build_opener(NoRedirect())
|
|
request = urllib.request.Request(
|
|
base + owner + "/" + repo,
|
|
headers={
|
|
"Accept": "application/json",
|
|
"User-Agent": "ZP-Agent repository verification",
|
|
},
|
|
)
|
|
with opener.open(request, timeout=4) as response:
|
|
raw = response.read(262145)
|
|
if len(raw) > 262144:
|
|
return "unverified"
|
|
data = json.loads(raw)
|
|
if (
|
|
data.get("private") is not False
|
|
or str(data.get("full_name", "")).casefold() != (owner + "/" + repo).casefold()
|
|
):
|
|
return "unverified"
|
|
# A repository can exist while a linked file/branch has been removed.
|
|
if len(parts) > 2:
|
|
with opener.open(
|
|
urllib.request.Request(
|
|
url, headers={"User-Agent": "ZP-Agent repository verification"}
|
|
),
|
|
timeout=4,
|
|
) as response:
|
|
if response.status != 200:
|
|
return "unverified"
|
|
return "verified"
|
|
except urllib.error.HTTPError as error:
|
|
code = error.code
|
|
error.close()
|
|
return "not_public" if code in {401, 403, 404, 410} else "unverified"
|
|
except Exception:
|
|
return "unverified"
|
|
|
|
|
|
def repository_status(url):
|
|
return _probe(url, int(time.monotonic() // 60))
|
|
|
|
|
|
def checked_repository_links(result, *, check=None):
|
|
if not result.get("accepted_for_delivery"):
|
|
return result
|
|
text = result.get("answer", "")
|
|
sources = result.get("source_urls", [])
|
|
urls = sorted(u for u in (text_urls(text) | set(sources)) if repository_reference(u))
|
|
if not urls:
|
|
return result
|
|
checker = check or repository_status
|
|
with ThreadPoolExecutor(max_workers=4) as pool:
|
|
statuses = dict(zip(urls[:8], pool.map(checker, urls[:8])))
|
|
bad = {url for url in urls if statuses.get(url) != "verified"}
|
|
if not bad:
|
|
return result
|
|
if bad & set(sources):
|
|
from tools.answers.answer_errors import public_failure
|
|
|
|
return public_failure({**result, "reason": "reference_unavailable"})
|
|
variants = {variant for url in bad for variant in (url, escape(url, quote=False))}
|
|
for url in sorted(variants, key=len, reverse=True):
|
|
text = re.sub(
|
|
r"\[([^]\n]+)\]\(\s*<?" + re.escape(url) + r'>?(?:\s+"[^"]*")?\s*\)', r"\1", text
|
|
)
|
|
text = re.sub(r"^\s*\[[^]\n]+\]:\s*<?" + re.escape(url) + r">?.*$", "", text, flags=re.M)
|
|
text = re.sub(
|
|
r'\s+(?:href|src)\s*=\s*(["\x27])([^"\x27]*)\1',
|
|
lambda m: "" if unescape(m[2]) == url else m[0],
|
|
text,
|
|
flags=re.I,
|
|
)
|
|
text = text.replace(url, "[odkaz neoverený]")
|
|
text += "\n\nRepozitárový odkaz zo zdroja sa nepodarilo verejne overiť, preto ho neuvádzam. "
|
|
text += "Nedostupnosť ani HTTP 404 samy nerozlišujú zmazaný a súkromný repozitár."
|
|
return {**result, "answer": text, "suppressed_repository_links": len(bad)}
|