"""Authoritative reference lookups for IT and software questions. Each of these replaces a class of confident-sounding hallucination: version numbers, RFC contents, CVE severity, command flags. They are worth more than their token cost precisely because the model cannot tell, from the inside, when its memory of a flag or a CVSS score is wrong. """ from __future__ import annotations import asyncio import html import logging import re from typing import Literal from fastapi import APIRouter, Depends from pydantic import Field from .. import cache from ..budget import paginate from ..httpc import get_bytes, get_json from ..security import require_auth from ..settings import settings from ..textwin import find_in_document, search_window from ..toolbase import NO_RESULTS, ToolModel, sources_line, tool log = logging.getLogger(__name__) router = APIRouter(prefix="/tools", tags=["reference"], dependencies=[Depends(require_auth)]) # ------------------------------------------------------- search_stackexchange SE_SITES = Literal[ "stackoverflow", "superuser", "serverfault", "unix", "electronics", "networkengineering", "security", "askubuntu", "raspberrypi", ] class StackExchangeRequest(ToolModel): query: str = Field( description="Search terms. Paste error messages verbatim - they match well.", ) site: SE_SITES = Field( default="stackoverflow", description="stackoverflow=programming (default); unix=Linux/shell; serverfault=servers " "and sysadmin; superuser=desktop computing; networkengineering=routing, " "switching, BGP; electronics=circuits, components, embedded; security=infosec; " "askubuntu=Ubuntu; raspberrypi=Pi.", ) max_results: int = Field(default=3, ge=1, le=6, description="Questions to return with their top answer, default 3, " "max 6.") @router.post("/search_stackexchange", operation_id="search_stackexchange", response_model=str, summary="Search Stack Exchange for a technical answer") @tool async def search_stackexchange(body: StackExchangeRequest) -> str: """Search Stack Overflow and its sister sites; returns questions with their best answers. Best for practical problems someone has already hit: error messages, "why does X fail", configuration that will not work, choosing between approaches, circuit questions on the electronics site. Prefer it over web_search for "how do I" and "why is this broken". Weigh the vote counts and acceptance shown, and check dates - a top answer from 2012 may describe software that has since changed. """ cache_key = (body.query, body.site, str(body.max_results)) if hit := await cache.cached("stackexchange", *cache_key): return hit base = "https://api.stackexchange.com/2.3" common = {"site": body.site} if settings.stackexchange_key: common["key"] = settings.stackexchange_key search = await get_json(f"{base}/search/advanced", params={ **common, "q": body.query, "sort": "relevance", "order": "desc", "pagesize": body.max_results, "filter": "withbody", }) items = search.get("items") or [] if not items: return NO_RESULTS blocks: list[str] = [] answer_ids = [str(q["accepted_answer_id"]) for q in items if q.get("accepted_answer_id")] answers: dict[int, dict] = {} if answer_ids: # One batched call for every accepted answer. The keyless quota is # 300 calls a day for the whole assistant, so fetching these one at # a time would be a meaningful waste of a shared budget. got = await get_json(f"{base}/answers/{';'.join(answer_ids[:6])}", params={ **common, "filter": "withbody", "sort": "votes", "order": "desc", }) for ans in got.get("items") or []: answers[ans["answer_id"]] = ans for question in items: title = html.unescape(question.get("title", "")) score = question.get("score", 0) link = question.get("link", "") answered = "accepted answer" if question.get("is_answered") else "no accepted answer" date = _epoch_to_date(question.get("creation_date")) block = [f"## {title}\nScore {score:+d} · {answered} · asked {date}\n{link}"] accepted = answers.get(question.get("accepted_answer_id") or -1) if accepted: text = _html_to_text(accepted.get("body", "")) block.append(f"\n**Accepted answer** (score {accepted.get('score', 0):+d}):\n{_clip(text, 1400)}") else: block.append(f"\n**Question**:\n{_clip(_html_to_text(question.get('body', '')), 600)}") block.append("\n[No accepted answer - treat any solution here as unverified.]") blocks.append("\n".join(block)) quota = search.get("quota_remaining") footer = "" if isinstance(quota, int) and quota < 25: footer = (f"\n\n[Only {quota} Stack Exchange lookups remain today across all users. " "Use them sparingly.]") result = "\n\n---\n\n".join(blocks) + footer await cache.store("stackexchange", result, *cache_key) return result def _epoch_to_date(epoch) -> str: import datetime if not epoch: return "unknown date" return datetime.datetime.fromtimestamp(epoch, datetime.UTC).strftime("%Y-%m-%d") def _html_to_text(raw: str) -> str: """Flatten Stack Exchange's answer HTML, keeping code blocks intact. Code is usually the whole point of the answer, so it survives as fenced blocks; everything else collapses to prose. """ if not raw: return "" text = re.sub(r"]*>]*>(.*?)", lambda m: f"\n```\n{html.unescape(m.group(1))}\n```\n", raw, flags=re.S) text = re.sub(r"]*>(.*?)", lambda m: f"`{html.unescape(m.group(1))}`", text, flags=re.S) text = re.sub(r"]*href=[\"']([^\"']+)[\"'][^>]*>(.*?)", r"\2 (\1)", text, flags=re.S) text = re.sub(r"]*>", "\n- ", text) text = re.sub(r"", "\n\n", text) text = re.sub(r"", "\n", text) text = re.sub(r"<[^>]+>", "", text) text = html.unescape(text) return re.sub(r"\n{3,}", "\n\n", text).strip() def _clip(text: str, limit: int) -> str: if len(text) <= limit: return text return text[:limit].rsplit("\n", 1)[0] + f"\n[… answer truncated at {limit} characters; open the link for the rest]" # ----------------------------------------------------------------- lookup_rfc class RfcRequest(ToolModel): number: int | None = Field( default=None, description="RFC number to read, e.g. 4271 for BGP-4.", ) query: str | None = Field( default=None, description="Keywords to find an RFC by topic when you do not know the number, e.g. " "'border gateway protocol'.", ) search: str | None = Field( default=None, description="Jump to a section or term - e.g. 'B.4', 'Cipher Suites', " "'TLS_AES_128_GCM_SHA256'. RFCs are long: use this for any specific section, " "table or constant rather than paging with `offset`.", ) offset: int = Field(default=0, ge=0, description="Character offset to continue a long RFC from.") # The datatracker index is a title search. A topic-shaped query - "DNS domain # name maximum length 63 octets label" - matches no title as a whole, and the # description promises a topic search, so the model asked exactly that and got # NO RESULTS with RFC 1035 one phrase away. Retry with the query's two-word # phrases, then its longer words, before giving up. _RFC_QUERY_STOP = frozenset( "the a an of in on for to and or is are was be by with from as at that this which what " "how does do did rfc rfcs spec specification standard define defines defined maximum " "minimum length size value values number numbers per each".split() ) async def _rfc_title_search(query: str) -> tuple[list[dict], str | None]: async def titles(term: str) -> list[dict]: found = await get_json("https://datatracker.ietf.org/api/v1/doc/document/", params={ "type": "rfc", "title__icontains": term, "format": "json", "limit": 8, }) return found.get("objects") or [] docs = await titles(query) if docs: return docs, None words = [w for w in re.findall(r"[A-Za-z][A-Za-z-]+", query) if w.lower() not in _RFC_QUERY_STOP and len(w) >= 3] candidates = [" ".join(pair) for pair in zip(words, words[1:])] candidates += [w for w in words if len(w) >= 5] for term in candidates[:6]: docs = await titles(term) if docs: return docs, term return [], None @router.post("/lookup_rfc", operation_id="lookup_rfc", response_model=str, summary="Read or search IETF RFCs") @tool async def lookup_rfc(body: RfcRequest) -> str: """Find an RFC by topic ("which RFC defines X"), or read one by number. The authoritative source for packet formats, state machines, header fields, timer defaults and reserved values. Use it whenever a networking question turns on a precise detail, and quote the specification rather than recalling it. Give `number` to read, or `query` to search by topic - including to check which RFC says something before citing an RFC number from memory. """ if body.number is None and not body.query: return "CANNOT DO THAT: provide either an RFC number or a search query." if body.number is None: docs, matched_on = await _rfc_title_search(body.query) if not docs: return (f"NO RESULTS. No RFC title contains '{body.query}' or any two-word phrase " "from it. This search matches words in RFC titles, not topics: name the " "protocol or subject the way its RFC title would ('Domain Names', 'HTTP " f"Semantics', 'Transport Layer Security'). {NO_RESULTS}") # Rank current standards above experimental and historic documents. # Searching "border gateway protocol" otherwise leads with RFC 1105, # the 1989 experimental BGP-1 - a correct hit and a badly wrong thing # to answer a BGP question from, since BGP-4 (RFC 4271) superseded it. docs.sort(key=lambda d: (_STATUS_RANK.get(_std_level(d), 5), -(d.get("rfc_number") or 0))) lines = ["RFCs matching that topic:\n" if matched_on is None else f"No RFC title contains the whole query; these titles contain '{matched_on}':\n"] for doc in docs: num = (doc.get("name") or "").removeprefix("rfc") abstract = re.sub(r"\s+", " ", doc.get("abstract") or "").strip() status = _STATUS_LABEL.get(_std_level(doc), "status unknown") year = (doc.get("time") or "")[:4] lines.append(f"- **RFC {num}** [{status}]: {doc.get('title', '')}\n {abstract[:260]}") lines.append( "\nRFCs are never edited - they are replaced. An older RFC on a topic has often " "been obsoleted by a newer one, and its status above tells you how much weight it " "carries. Prefer an Internet Standard or Proposed Standard over an Experimental or " "Historic document, and if you are unsure whether one is current, read it and check " "its header for an 'Obsoleted by' line before citing it." ) lines.append( "\nThese are index entries: title, publication status and abstract. The body of an " "RFC - its sections, field definitions, reserved values and exact wording - is not " "in this listing, so nothing can be quoted from it. To read one: lookup_rfc with " "number=, or number= and search='
'." ) return "\n".join(lines) number = body.number cache_key = (str(number),) text = await cache.cached("rfc", *cache_key) if text is None: raw, _, _ = await get_bytes(f"https://www.rfc-editor.org/rfc/rfc{number}.txt") if not raw: return f"NO RESULTS. RFC {number} could not be retrieved. Check the number is right." text = _strip_rfc_pagination(raw.decode("utf-8", errors="replace")) await cache.store("rfc", text, *cache_key) header = f"# RFC {number}\nSource: https://www.rfc-editor.org/rfc/rfc{number}.txt\n\n" trailer = sources_line(f"RFC {number} - https://www.rfc-editor.org/rfc/rfc{number}.txt") if body.search: return header + search_window( text, body.search, window=5000, continue_hint=f"call lookup_rfc with number={number} and offset= to read more", miss_note=(f"'{body.search}' does not appear in RFC {number}. It may be defined in a " "different RFC, or named differently here - do not state it as this RFC's " "content."), ) + trailer return header + paginate( text, body.offset, 7000, how_to_continue=f"To continue, call lookup_rfc with number={number} and offset={{offset}}, " f"or with search='
' to jump straight to it.", ) + trailer def _strip_rfc_pagination(text: str) -> str: """Remove the fixed-width form feeds and running headers. Left in, they interleave "[Page 12]" and a repeated title banner into the middle of sentences every 58 lines, which reads to a model as though the specification says something it does not. """ out: list[str] = [] for line in text.replace("\f", "\n\f\n").splitlines(): stripped = line.strip() if stripped.startswith("\f"): continue if re.match(r"^RFC \d+\s+.*\s+\w+ \d{4}$", stripped): continue if re.search(r"\[Page \d+\]$", stripped): continue out.append(line.rstrip()) return re.sub(r"\n{3,}", "\n\n", "\n".join(out)).strip() # std_level arrives as a URI like /api/v1/name/stdlevelname/ps/; the last # segment is the code. Extracting it costs nothing, whereas resolving the # obsoletes/obsoleted-by relationships would be one extra API call per hit. _STATUS_LABEL = { "std": "Internet Standard", "ds": "Draft Standard", "ps": "Proposed Standard", "bcp": "Best Current Practice", "inf": "Informational", "exp": "Experimental", "hist": "Historic", "unkn": "status unknown", } _STATUS_RANK = {"std": 0, "bcp": 0, "ds": 1, "ps": 1, "inf": 3, "exp": 4, "hist": 4} def _std_level(doc: dict) -> str: return (doc.get("std_level") or "").rstrip("/").rsplit("/", 1)[-1].lower() # ----------------------------------------------------------------- lookup_cve class CveRequest(ToolModel): cve_id: str | None = Field( default=None, description="Exact identifier, e.g. CVE-2021-44228.", ) keyword: str | None = Field( default=None, description="Product or technology to search, e.g. 'openssh' or 'cisco ios xe'.", ) @router.post("/lookup_cve", operation_id="lookup_cve", response_model=str, summary="Look up a CVE or search vulnerabilities") @tool async def lookup_cve(body: CveRequest) -> str: """Authoritative details for a CVE: description, CVSS severity, affected versions, CISA KEV status. Use it for any question about a specific CVE; never state a CVSS score or an affected version from memory. Give `cve_id` for one vulnerability, or `keyword` to search a product. If the CVE is not found, say so rather than describing what it might be from the number alone. """ if not body.cve_id and not body.keyword: return "CANNOT DO THAT: provide either a CVE ID or a keyword to search for." if body.cve_id: cve_id = body.cve_id.strip().upper() if not re.match(r"^CVE-\d{4}-\d{4,}$", cve_id): return f"CANNOT DO THAT: '{body.cve_id}' is not a valid CVE identifier (expected CVE-YYYY-NNNNN)." if hit := await cache.cached("cve", cve_id): return hit # CIRCL aggregates the CNA record with NVD enrichment and has no hard # rate limit, unlike NVD's documented 5-requests-per-30-seconds. record = None for url in (f"https://cve.circl.lu/api/cve/{cve_id}", f"https://cveawg.mitre.org/api/cve/{cve_id}"): try: record = await get_json(url) if record: break except Exception as exc: log.info("CVE source %s failed: %s", url, exc) if not record: return (f"NO RESULTS. {cve_id} was not found in any vulnerability database. " "It may not exist, or may be too recently reserved to have details. " "Do not describe it from the identifier alone.") # Exploitation status and CPE version ranges are not in the CVE record # at all; they are in NVD's copy, which the keyword branch below already # reads. Fetched here so both land in the same cached result. result = _format_cve(cve_id, record, await _nvd_cve(cve_id)) await cache.store("cve", result, cve_id) return result # Keyword search: NVD is the only one of the three with real free-text search. params = {"keywordSearch": body.keyword, "resultsPerPage": 5} headers = {"apiKey": settings.nvd_key} if settings.nvd_key else {} payload = await get_json("https://services.nvd.nist.gov/rest/json/cves/2.0", params=params, headers=headers, timeout=30.0) vulns = payload.get("vulnerabilities") or [] if not vulns: return f"NO RESULTS. No vulnerabilities found matching '{body.keyword}'." total = payload.get("totalResults", len(vulns)) lines = [f"{total} vulnerabilities match '{body.keyword}'. Showing the {len(vulns)} most recent:\n"] for entry in vulns: cve = entry.get("cve", {}) cid = cve.get("id", "?") desc = next((d["value"] for d in cve.get("descriptions", []) if d.get("lang") == "en"), "") lines.append(f"- **{cid}** ({_nvd_severity(cve)}): {re.sub(r'\\s+', ' ', desc)[:220]}") lines.append("\nNext: call lookup_cve with a specific cve_id for full details.") return "\n".join(lines) # CVSS lives in one of two places, and reading only the first one is wrong. # # A CVE record has a `cna` container (from whoever assigned the CVE) and zero or # more `adp` containers (Authorized Data Publishers - in practice CISA-ADP, which # back-fills scoring the CNA never supplied). For older CVEs the CNA container # routinely has metrics: null while the ADP container carries the real score. # # Reading only `cna` reported "No CVSS score has been assigned" for CVE-2014-0160 # - Heartbleed - which NVD scores 7.5 HIGH. The assistant caught that and # corrected it from the web, but it cost three extra tool calls and a paragraph # of hedging on a question the tool should simply have answered. _CVSS_KEYS = ( ("cvssV4_0", "4.0"), ("cvssV3_1", "3.1"), ("cvssV3_0", "3.0"), ("cvssV2_0", "2.0"), ) def _cvss_line(record: dict) -> str: containers = record.get("containers") or {} sources: list[tuple[str, list]] = [] cna = containers.get("cna") or {} sources.append(("CNA", cna.get("metrics") or [])) for adp in containers.get("adp") or []: if isinstance(adp, dict): name = ((adp.get("providerMetadata") or {}).get("shortName") or "ADP") sources.append((name, adp.get("metrics") or [])) # (version, score, severity, vector) -> the sources that published it. scores: dict[tuple, list[str]] = {} for label, metrics in sources: for metric in metrics: if not isinstance(metric, dict): continue for key, version in _CVSS_KEYS: data = metric.get(key) if not isinstance(data, dict) or data.get("baseScore") is None: continue ident = (version, data.get("baseScore"), data.get("baseSeverity", ""), data.get("vectorString", "")) if label not in scores.setdefault(ident, []): scores[ident].append(label) if not scores: return ("No CVSS score is published in the CVE record. Do not infer one from the " "description - say it is unscored, or check NVD directly.") order = {v: i for i, (_, v) in enumerate(_CVSS_KEYS)} ranked = sorted(scores, key=lambda ident: order.get(ident[0], 9)) def render(ident: tuple) -> str: version, score, severity, vector = ident return f"CVSS {version}: {score} ({severity}) {vector}".replace(" ", " ").strip() # Name the publisher ONLY where publishers actually disagree. Tagging a lone # score with its source reads to the model as "one of several competing # numbers" and it will invent the disagreement - see Change 2 in TUNING.md. same_version = [i for i in ranked if i[0] == ranked[0][0]] if len(same_version) > 1: lines = ["Scoring sources disagree - report the disagreement, do not pick one silently:"] lines += [f" {render(i)} [per {', '.join(scores[i])}]" for i in same_version[:4]] return "\n".join(lines) primary = render(ranked[0]) older = [render(i) for i in ranked[1:3]] return primary if not older else primary + "\n" + "\n".join( f"Also published: {o}" for o in older) def _nvd_severity(cve: dict) -> str: metrics = cve.get("metrics") or {} for key in ("cvssMetricV31", "cvssMetricV30", "cvssMetricV2"): if entries := metrics.get(key): data = entries[0].get("cvssData", {}) score = data.get("baseScore") sev = data.get("baseSeverity") or entries[0].get("baseSeverity") or "" if score is not None: return f"CVSS {score} {sev}".strip() return "no score assigned" async def _nvd_cve(cve_id: str) -> dict | None: """NVD's own record for one CVE, or None when it could not be read. Two things worth an extra call live here and nowhere in the CVE record: whether CISA has put the CVE in the Known Exploited Vulnerabilities catalogue, and NVD's CPE version intervals. Every failure returns None rather than raising. NVD throttles keyless callers to five requests per thirty seconds, and a rate-limited enrichment lookup must not take down the CVE record, which is the actual answer. """ headers = {"apiKey": settings.nvd_key} if settings.nvd_key else {} try: payload = await get_json("https://services.nvd.nist.gov/rest/json/cves/2.0", params={"cveId": cve_id}, headers=headers, timeout=30.0) entries = payload.get("vulnerabilities") or [] return (entries[0].get("cve") or None) if entries else None except Exception as exc: log.info("NVD enrichment for %s failed: %s", cve_id, exc) return None def _kev_line(nvd: dict | None) -> str: """Whether CISA lists the CVE as exploited in the wild. This is the one field that decides whether something is patched tonight or at the next maintenance window, and it is not derivable from the description - the model cannot tell an exploited bug from a theoretical one by reading about it. The unavailable branch is stated rather than omitted, because a missing line reads as "not listed". """ if nvd is None: return ("**CISA KEV:** status unavailable (the NVD lookup failed) - do not state " "whether it is listed.") added = nvd.get("cisaExploitAdd") if not added: return "**CISA KEV:** not listed as of this lookup." due = nvd.get("cisaActionDue") or "unstated" name = nvd.get("cisaVulnerabilityName") or "" return (f"**CISA KEV:** listed - added {added}, remediation due {due}" f"{f' ({name})' if name else ''}. This is known to be exploited in the wild.") # CPE intervals are expressed as four optional bounds, at most one of each end. _CPE_BOUNDS = ( ("versionStartIncluding", "from {}"), ("versionStartExcluding", "after {}"), ("versionEndIncluding", "through {}"), ("versionEndExcluding", "before {}"), ) def _cpe_ranges(nvd: dict | None, limit: int = 6) -> list[str]: """NVD's machine-readable version intervals, deduplicated and capped. The CNA's own `affected` rows are frequently vaguer than this - a product with no versions at all, or one build number where a whole branch is meant - while the CPE data always carries an interval. A "which versions do I have to patch" question is answered from here when both are present. The cap has to pick which rows to keep, and document order is the wrong rule: CVE-2021-44228 has 396 rows of which the first six are Siemens firmware images, with Apache log4j itself hundreds of rows down. Rows for a product the description actually names come first, so a widely-embedded library is reported as itself rather than as six of its downstream appliances. Frequency does not work here - Cisco contributes more rows to log4shell than log4j does. """ if not nvd: return [] described = " ".join(d.get("value") or "" for d in (nvd.get("descriptions") or [])).lower() ranked: list[tuple[int, int, str]] = [] seen: set[str] = set() for config in nvd.get("configurations") or []: for node in config.get("nodes") or []: for match in node.get("cpeMatch") or []: if not isinstance(match, dict) or not match.get("vulnerable", True): continue parts = (match.get("criteria") or "").split(":") if len(parts) < 6: continue product = parts[4].replace("_", " ") if parts[4] not in ("*", "-", "") else "" name = " ".join(p.replace("_", " ") for p in parts[3:5] if p not in ("*", "-", "")) bounds = [shape.format(match[key]) for key, shape in _CPE_BOUNDS if match.get(key)] if not bounds: if parts[5] in ("*", "-", ""): continue bounds = [parts[5]] line = f"{name}: {' '.join(bounds)}" if name else " ".join(bounds) if line in seen: continue seen.add(line) ranked.append((0 if product and product in described else 1, len(ranked), line)) ranked.sort() return [line for _, _, line in ranked[:limit]] # Names that are present in the JSON but say nothing. CVE records use the # literal string "n/a" for "this product has no vendor" as often as they use # null, and both used to print. _NO_NAME = {"", "n/a", "n\\/a", "unknown", "unspecified", "not applicable", "*", "-", "?"} # Version strings meaning "from the beginning", which read as a real lower # bound if printed: "from 0 before 116.0.5845.187". _VERSION_ANY = {"0", "0.0", "*"} _VERSION_NONE = {"", "-", "n/a", "unknown", "unspecified"} def _plain(value) -> str: """A usable name, or "" for anything that is not one. `item.get("vendor", "?")` supplies the default only when the key is absent, so an explicit JSON null - which is how a CNA says "no vendor" - printed "None Chrome", and a record with both fields null printed "- ? ?: ...", a bullet point that is not a fact about anything. """ if not isinstance(value, str): return "" value = value.strip() return "" if value.lower() in _NO_NAME else value def _affected_name(item: dict) -> str: """Name one affected thing, or "" when the record does not name it. Package ecosystems fill in `packageName` and leave vendor and product null, so an openssh advisory with a perfectly good name rendered as "? ?". """ vendor = _plain(item.get("vendor")) product = _plain(item.get("product")) or _plain(item.get("packageName")) named = " ".join(part for part in (vendor, product) if part) return named or _plain(item.get("collectionURL")) def _version_text(entry: dict) -> str: """Render one `versions` row with its bounds. A row meaning "everything below 116.0.5845.187 is affected" is {"version": "0", "lessThan": "116.0.5845.187", "status": "affected"}. Reading only `version` and `status` printed "116.0.5845.187 (affected)" - the single version the row excludes, stated as the one version it names. That inversion went out as an answer (m01), and it is the worst possible error on this tool: someone reads it as "upgrade to 116.0.5845.187 is not enough" or as "this build is the vulnerable one", both backwards. """ version = _vstr(entry.get("version")) below = _vstr(entry.get("lessThan")) upto = _vstr(entry.get("lessThanOrEqual")) lower = "" if version in _VERSION_ANY else version if below: text = f"from {lower} before {below}" if lower else f"before {below}" elif upto: text = f"from {lower} through {upto}" if lower else f"through {upto}" elif version == "*": text = "all versions" else: text = version if not text: return "" if status := _plain(entry.get("status")): text += f" ({status})" if (entry.get("versionType") or "").strip().lower() == "git": # The bounds are commit ids. Read as release numbers they name versions # that were never shipped, and "fixed in 4de3b2..." is not something a # user can check against what they are running. text += " [a git commit range, not release numbers]" return text def _vstr(value) -> str: if not isinstance(value, str): return "" value = value.strip() return "" if value.lower() in _VERSION_NONE else value def _affected_rows(record: dict) -> list[dict]: """The affected list, from the ADP container when the CNA supplied none. Same split as the CVSS scores above: for older or thinly-documented CVEs the CNA container carries no `affected` at all while CISA-ADP's does, and reading only `cna` printed no Affected section for a record that states the versions plainly. No source tag is attached - see the note on _cvss_line; naming a lone source is what got amplified into an invented disagreement. """ containers = record.get("containers") or {} rows = ((containers.get("cna") or {}).get("affected")) or [] if rows: return rows for adp in containers.get("adp") or []: if isinstance(adp, dict) and (rows := adp.get("affected") or []): return rows return [] def _format_cve(cve_id: str, record: dict, nvd: dict | None = None) -> str: container = (record.get("containers") or {}).get("cna") or {} meta = record.get("cveMetadata") or {} descriptions = container.get("descriptions") or [] summary = next((d.get("value") for d in descriptions if d.get("lang", "").startswith("en")), "") if not summary: summary = record.get("summary") or "(no description published)" lines = [f"# {cve_id}", ""] if state := meta.get("state"): lines.append(f"State: {state}") if published := (meta.get("datePublished") or record.get("Published") or ""): lines.append(f"Published: {published[:10]}") # Severity, exploitation and version ranges go above the description # because cap() trims this response from the end, and a truncated answer # that kept the prose and dropped "known to be exploited in the wild" is # the wrong half to keep. lines += ["", f"**Severity:** {_cvss_line(record)}", "", _kev_line(nvd)] if ranges := _cpe_ranges(nvd): lines += ["", "**Affected versions (NVD):**"] + [f"- {r}" for r in ranges] lines += ["", "## Description", re.sub(r"\s+", " ", summary)] affected = _affected_rows(record) if affected: lines += ["", "## Affected (as published in the CVE record)"] for item in affected[:6]: name = _affected_name(item) versions = "; ".join( text for text in (_version_text(v) for v in (item.get("versions") or [])[:4]) if text ) if name and versions: lines.append(f"- {name}: {versions}") elif name or versions: lines.append(f"- {name or versions}") refs = [r.get("url") for r in (container.get("references") or [])[:5] if r.get("url")] if refs: lines += ["", "## References"] + [f"- {u}" for u in refs] return "\n".join(lines) # -------------------------------------------------------------- read_man_page class ManPageRequest(ToolModel): name: str = Field( description="Command or function name, e.g. 'ip', 'tcpdump', 'systemd.network'. Multi-word " "commands work as typed: 'ip route' reaches ip-route, 'git commit' reaches " "git-commit; where no such page exists, the first word's page is read at the " "second word.", ) section: int | None = Field( default=None, description="Manual section: 1=commands, 2=syscalls, 3=library functions, 4=devices, " "5=file formats, 7=overviews, 8=admin commands. Empty means auto-resolve.", ) search: str | None = Field( default=None, description="Jump to a flag or keyword - e.g. '-Z', '--relinquish-privileges', 'EXAMPLES'. " "Preferred when you want one option out of a long page.", ) offset: int = Field(default=0, ge=0, description="Character offset to continue a long page " "from.") @router.post("/read_man_page", operation_id="read_man_page", response_model=str, summary="Read a Linux manual page") @tool async def read_man_page(body: ManPageRequest) -> str: """Read the manual page for a Linux/Unix command, syscall or config file. Use it before describing any command's flags or a config file's syntax - options change between versions, and a wrong flag lands in the user's terminal. Read the page and quote it. """ words = body.name.split() # _man_name used to run on the raw name, and its character filter simply # deleted the space: "ip route" became "iproute", the fetch 404ed, and the # tool answered "No manual page found for 'ip route' ... Do not describe its # options from memory" - a flat false negative, since ip-route(8) exists. # Manual pages for subcommands are named with a hyphen, so that is what a # space means here. page = _man_name("-".join(words)) if not page: return "CANNOT DO THAT: provide a command or function name." slug = f"{page}.{body.section}" if body.section else page text = await _man_page(slug) note = "" missing = "" search = body.search if text is None and len(words) > 1: # No hyphenated page for the phrase. Many subcommands are documented as # a section inside the parent page instead ("systemctl enable" lives in # systemctl(1)), so one retry there answers the question that was asked # rather than reporting the command does not exist. first = _man_name(words[0]) alt_slug = f"{first}.{body.section}" if body.section else first alt = await _man_page(alt_slug) if alt is not None: search = search or " ".join(words[1:]) missing = slug note = f"[No page '{missing}'; showing '{first}' at '{search}'.]" page, slug, text = first, alt_slug, alt if text is None: return (f"NO RESULTS. No manual page found for '{body.name}'" f"{f' in section {body.section}' if body.section else ''}. " "Check the spelling, or the command may not be packaged for Linux. " "Do not describe its options from memory.") if not text.strip(): return f"NO RESULTS. The manual page for '{body.name}' is empty." # A bare name resolves to the lowest section, and the fact asked about # is often in another one: cron's default PATH is documented in # crontab(5), the file-format page, while a search of "crontab" landed # in crontab(1) and reported the term absent. When the term misses and # no section was given, the sibling pages are checked before answering. sibling_note = "" if search and body.section is None and find_in_document(text, search) is None: current = _section_of(text) candidates = [sec for sec in ("5", "8", "1", "7", "3", "2") if sec != current] pages = await asyncio.gather(*(_man_page(f"{page}.{sec}") for sec in candidates)) others = [(sec, alt) for sec, alt in zip(candidates, pages) if alt is not None] hit = next(((sec, alt) for sec, alt in others if find_in_document(alt, search)), None) if hit: sec, alt = hit note = (f"[No '{search}' in {slug} ({_SECTION_NAMES.get(current, 'section ' + (current or '?'))}). " f"Showing {page}.{sec} ({_SECTION_NAMES.get(sec, 'section ' + sec)}), which does " "contain it.]") slug, text, body.section = f"{page}.{sec}", alt, int(sec) if sec.isdigit() else None elif others: sibling_note = (f" This is {slug} ({_SECTION_NAMES.get(current, 'section ' + (current or '?'))}); " f"the name also exists in " + ", ".join(f"section {sec} ({_SECTION_NAMES.get(sec, '')})" for sec, _ in others) + " - pass section= to read one of those.") where = f"name='{page}'" + (f" and section={body.section}" if body.section else "") trailer = sources_line(f"man {slug} - https://man.archlinux.org/man/{slug}.txt") if search: # The miss path used to return the notice alone; it now shows the page # anyway, which is what the retry branch below had already concluded is # the right behaviour. A term that is not on a page is not the same # finding as the page being useless. return _man_header(slug, note) + search_window( text, search, continue_hint=f"call read_man_page with {where} and offset= to read more", miss_note=(f"'{search}' does not appear in this manual page. It may be spelled " "differently, belong to a different command, or not exist - check before " f"describing it.{sibling_note}"), ) + trailer return _man_header(slug, note) + paginate( text, body.offset, 7000, how_to_continue=f"To continue, call read_man_page with {where} and offset={{offset}}, " f"or with search='' to jump straight to one option.", ) + trailer def _man_name(raw: str) -> str: return re.sub(r"[^A-Za-z0-9_.+-]", "", raw.strip()) _SECTION_NAMES = { "1": "user commands", "2": "system calls", "3": "library functions", "4": "devices", "5": "file formats and configuration", "6": "games", "7": "overviews and conventions", "8": "administration commands", } def _section_of(text: str) -> str | None: """The section a rendered page belongs to, from its 'NAME(1)' banner.""" for line in text.splitlines()[:5]: if found := re.search(r"\S\((\d)[A-Za-z]*\)", line): return found.group(1) return None def _man_header(slug: str, note: str = "") -> str: return (f"# man {slug}\nSource: https://man.archlinux.org/man/{slug}.txt\n" + (note + "\n" if note else "") + "\n") async def _man_page(slug: str) -> str | None: """Return the rendered page, or None when man.archlinux.org has no such page. Separate from the tool body so the multi-word retry can ask for a second page without duplicating the cache handling. """ text = await cache.cached("man", slug) if text is not None: return text try: raw, _, _ = await get_bytes(f"https://man.archlinux.org/man/{slug}.txt") except Exception: return None text = raw.decode("utf-8", errors="replace") if text.strip(): await cache.store("man", text, slug) return text # ------------------------------------------------------------ lookup_package # Suites still receiving updates, plus the rolling names. Anything else - # etch, sarge, woody, jessie, stretch, buster - is an archive entry, not an # answer to "what version does Debian have". _CURRENT_DEBIAN_SUITES = { "sid", "unstable", "testing", "experimental", "stable", "oldstable", "forky", "trixie", "bookworm", "bullseye", } def _is_current_suite(suite: str) -> bool: return suite.split("-", 1)[0].lower() in _CURRENT_DEBIAN_SUITES ECOSYSTEMS = Literal["pypi", "npm", "crates", "dockerhub", "github", "debian"] class PackageRequest(ToolModel): name: str = Field( description="Package name. For github, 'owner/repo'. For dockerhub, 'nginx' for official " "images, 'org/image' otherwise.", ) ecosystem: ECOSYSTEMS = Field( description="pypi=Python, npm=Node, crates=Rust, dockerhub=container images, " "github=releases and tags, debian=Debian/Ubuntu packages.", ) @router.post("/lookup_package", operation_id="lookup_package", response_model=str, summary="Get the current version of a package") @tool async def lookup_package(body: PackageRequest) -> str: """Current released version of a software package, and its release date. Use it instead of ever answering "the latest version is..." from memory - versions move constantly. It also confirms a package exists before you recommend it. """ name = body.name.strip() if not name: return "CANNOT DO THAT: provide a package name." cache_key = (body.ecosystem, name.lower()) if hit := await cache.cached("package", *cache_key): # Appended after the cache, not before it: the cached value is the # registry's answer, and the citation line is built from it. return hit + _package_sources(hit) try: result = await _fetch_package(body.ecosystem, name) except Exception as exc: log.info("package lookup failed for %s/%s: %s", body.ecosystem, name, exc) return (f"NO RESULTS. '{name}' was not found on {body.ecosystem}. Check the exact name " "and spelling. Do not state a version you are not sure of.") await cache.store("package", result, *cache_key) return result + _package_sources(result) def _package_sources(result: str) -> str: """The registry page for whatever was just looked up. The version and the minimum runtime came from a registry with a URL, and across two graded rounds the answer never carried it - the `Homepage:` row is one field among five and reads as data about the package, not as a source for the claim. This is the same fact, shaped like a citation. """ if match := re.search(r"^Homepage: (\S+)$", result, re.MULTILINE): return sources_line(match.group(1)) return "" async def _fetch_package(ecosystem: str, name: str) -> str: if ecosystem == "pypi": data = await get_json(f"https://pypi.org/pypi/{name}/json") info = data["info"] releases = data.get("releases", {}).get(info["version"], []) when = releases[0].get("upload_time", "")[:10] if releases else "unknown" return (f"# {info['name']} (PyPI)\n\n**Latest version: {info['version']}** " f"(released {when})\n\n{info.get('summary', '')}\n\n" f"Requires Python: {info.get('requires_python') or 'unspecified'}\n" f"License: {info.get('license') or 'unspecified'}\n" f"Homepage: {info.get('home_page') or info.get('project_url', '')}\n" f"Install: pip install {info['name']}") if ecosystem == "npm": # /latest rather than the full document: the full one is ~340KB of # every version ever published, for one version number. data = await get_json(f"https://registry.npmjs.org/{name}/latest") return (f"# {data.get('name')} (npm)\n\n**Latest version: {data.get('version')}**\n\n" f"{data.get('description', '')}\n\nLicense: {data.get('license', 'unspecified')}\n" f"Homepage: {data.get('homepage', '')}\nInstall: npm install {data.get('name')}") if ecosystem == "crates": data = await get_json(f"https://crates.io/api/v1/crates/{name}") crate = data["crate"] return (f"# {crate['name']} (crates.io)\n\n" f"**Latest stable version: {crate.get('max_stable_version')}** " f"(newest including pre-release: {crate.get('newest_version')})\n" f"Updated: {(crate.get('updated_at') or '')[:10]}\n\n" f"{crate.get('description', '')}\n\nDownloads: {crate.get('downloads', 0):,}\n" f"Install: cargo add {crate['name']}") if ecosystem == "dockerhub": repo = name if "/" in name else f"library/{name}" data = await get_json(f"https://hub.docker.com/v2/repositories/{repo}/tags", params={"page_size": 8, "ordering": "last_updated"}) tags = data.get("results") or [] if not tags: raise ValueError("no tags") lines = [f"# {name} (Docker Hub)\n", f"{data.get('count', 0)} tags total. Most recently updated:\n"] for tag in tags: size = tag.get("full_size") or 0 lines.append(f"- `{tag['name']}` — updated {(tag.get('last_updated') or '')[:10]}" f"{f', {size / 1048576:.0f} MB' if size else ''}") return "\n".join(lines) if ecosystem == "github": if "/" not in name: return "CANNOT DO THAT: GitHub lookups need 'owner/repo' format." headers = {"Accept": "application/vnd.github+json"} if settings.github_token: headers["Authorization"] = f"Bearer {settings.github_token}" try: rel = await get_json(f"https://api.github.com/repos/{name}/releases/latest", headers=headers) return (f"# {name} (GitHub)\n\n**Latest release: {rel.get('tag_name')}** " f"({(rel.get('published_at') or '')[:10]})\n\n" f"{rel.get('name', '')}\n\n{(rel.get('body') or '')[:900]}\n\n{rel.get('html_url', '')}") except Exception: # A repo that tags but never publishes releases 404s here - which # is not the same as the repo being missing. torvalds/linux is the # canonical example. tags = await get_json(f"https://api.github.com/repos/{name}/tags", params={"per_page": 5}, headers=headers) if not tags: raise listed = "\n".join(f"- {t['name']}" for t in tags) return (f"# {name} (GitHub)\n\nThis repository publishes tags rather than releases. " f"Most recent tags:\n{listed}\n\nhttps://github.com/{name}") if ecosystem == "debian": data = await get_json(f"https://sources.debian.org/api/src/{name}/") versions = data.get("versions") or [] if not versions: raise ValueError("no versions") # sources.debian.org returns every suite ever, newest-first by its own # ordering but with no notion of which are still shipping. Taking the # first eight answered "what does Debian have?" for `postgresql` with # six releases from etch and earlier - 7.5.22, 7.4.7, 6.5.3 - presented # as if current. The real answer is that this source name is not in any # supported suite at all, because Debian ships versioned names now. current, historical = [], [] for entry in versions: suites = entry.get("suites") or [] if any(_is_current_suite(x) for x in suites): current.append(entry) else: historical.append(entry) lines = [f"# {name} (Debian)\n"] if current: lines.append("In currently supported suites:\n") for entry in current[:8]: lines.append(f"- **{entry.get('version')}** — {', '.join(entry.get('suites') or [])}") else: newest = historical[0] if historical else None lines += [ "**Not present in any currently supported Debian suite.**\n", f"The newest entry for this source package is {newest.get('version')} in " f"{', '.join(newest.get('suites') or [])}, which is long out of support." if newest else "", "", "Debian often ships versioned source names - postgresql-17, llvm-toolchain-19, " f"gcc-14 - so try '{name}-', or check the link below for what the " "binary package actually maps to. Do NOT report the versions below as what " "Debian provides today.", ] if historical and current: lines.append(f"\n({len(historical)} older entries in unsupported suites omitted.)") elif historical and not current: lines.append("\nHistorical entries: " + ", ".join( f"{e.get('version')} ({', '.join(e.get('suites') or [])})" for e in historical[:5])) lines.append(f"\nhttps://packages.debian.org/{name}") return "\n".join(x for x in lines if x != "") raise ValueError(f"unsupported ecosystem {ecosystem}")