#!/usr/bin/env python3 """ Maven MCP server — MCP stdio (JSON-RPC 2.0) over stdin/stdout. No external dependencies; Python 3.9+ standard library only. """ import base64 import concurrent.futures import datetime import email.utils import functools import hashlib import http.client import http.server import ipaddress import json import logging import os import random import re import socket import ssl import subprocess import sys import tempfile import threading import time import urllib.request import urllib.parse import urllib.error import zlib from typing import Any, Callable, Dict, List, Optional, Set, Tuple # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- SERVER_NAME = "maven-mcp" SERVER_VERSION = "1.1.0" USER_AGENT = "maven-mcp/1.1.0" # MCP protocol revisions this server negotiates (#398). Every message shape this # server emits (tool annotations, outputSchema/structuredContent) is additive-only # across these revisions -- nothing here is gated by which one was negotiated -- so # "supported" means "willing to echo back", per the spec's version-negotiation rule: # "If the server supports the requested protocol version, it MUST respond with the # same version. Otherwise ... another protocol version it supports ... SHOULD be the # latest version supported by the server." 2024-11-05 is kept in this set for # clients still pinned to it (a real-world MCP client that only ever requests an # older revision must get that exact revision back, not be forced onto a newer one # it may not recognise and then disconnect per spec). MCP_SUPPORTED_PROTOCOL_VERSIONS = ("2025-11-25", "2025-06-18", "2025-03-26", "2024-11-05") MCP_LATEST_PROTOCOL_VERSION = "2025-11-25" HTTP_TIMEOUT = 15 # Short timeout for enrichment APIs (OSV / GitHub / deps.dev / android.com). # Closed contours must fail fast rather than hang on the full HTTP_TIMEOUT (#296). HTTP_TIMEOUT_EXTERNAL = 3 # Bounded retry/backoff (ported from the retired TS `fetchWithRetry`). The TS # default was `retries = 1` => 2 total attempts; this matches it. Backoff base # mirrors the TS 200ms + up-to-300ms jitter, here applied exponentially. HTTP_MAX_ATTEMPTS = 2 # total attempts (1 initial + 1 retry) HTTP_BACKOFF_BASE = 0.2 # seconds; delay = base * 2**attempt + jitter HTTP_BACKOFF_JITTER = 0.3 # seconds; uniform [0, jitter) added per attempt HTTP_RETRY_AFTER_MAX = 30.0 # cap an upstream Retry-After so a hostile header can't stall us HTTP_TOTAL_RETRY_BUDGET = 60.0 # wall-clock cap across all attempts + sleeps # Cap on a single HTTP response body. Maven metadata/POMs are KB-scale; # OSV/GitHub/Solr JSON is typically well under a few MB. A hostile or # misconfigured endpoint returning multi-GB bodies must not OOM the # long-lived stdio server (#350). HTTP_TIMEOUT only bounds time, not size. HTTP_MAX_RESPONSE_BYTES = 10 * 1024 * 1024 # 10 MiB MAVEN_CENTRAL_URL = "https://repo1.maven.org/maven2" GOOGLE_MAVEN_URL = "https://dl.google.com/dl/android/maven2" GRADLE_PLUGIN_PORTAL_URL = "https://plugins.gradle.org/m2" GOOGLE_MAVEN_GROUPS = ( "androidx.", "com.google.android.", "com.android.", "com.google.firebase.", "com.google.gms.", "com.google.mlkit.", "com.google.ar.", ) STABILITY_PATTERNS = [ (re.compile(r'[-.]?SNAPSHOT$', re.IGNORECASE), "snapshot"), (re.compile(r'[-.](?:alpha|a(?=\d|[-.]|$))[-.]?\d*', re.IGNORECASE), "alpha"), (re.compile(r'[-.](?:beta|b(?=\d|[-.]|$))[-.]?\d*', re.IGNORECASE), "beta"), (re.compile(r'[-.](?:M|milestone)[-.]?\d*', re.IGNORECASE), "milestone"), (re.compile(r'[-.](?:RC|CR)[-.]?\d*', re.IGNORECASE), "rc"), ] PRERELEASE_WEIGHT = {"snapshot": 0, "alpha": 1, "beta": 2, "milestone": 3, "rc": 4, "stable": 5} # Default public enrichment endpoints. Overridable via MAVEN_MCP_*_BASE (#296). GITHUB_API_DEFAULT = "https://api.github.com" OSV_API_DEFAULT = "https://api.osv.dev/v1/querybatch" OSV_VULN_API_DEFAULT = "https://api.osv.dev/v1/vulns" DEPSDEV_API_DEFAULT = "https://api.deps.dev/v3" # endoflife.date v1 API (#415) — JDK/Kotlin/Gradle/Spring Boot EOL/support status. _ENDOFLIFE_API_DEFAULT = "https://endoflife.date/api/v1" # Back-compat aliases — tests and docs historically referenced these names. # Prefer the *_url() / *_base() resolvers at call sites so env overrides apply. GITHUB_API = GITHUB_API_DEFAULT OSV_API = OSV_API_DEFAULT OSV_VULN_API = OSV_VULN_API_DEFAULT # OSV.dev documents a maximum of 1000 queries per /v1/querybatch request. # query_osv_batch chunks above this so a large monorepo audit cannot fail as a unit. OSV_QUERYBATCH_MAX = 1000 # /v1/querybatch returns only {id, modified} per vuln — severity/summary/fixed # require a follow-up GET /v1/vulns/{id} (#338). Cap bounds worst-case fan-out # per query_osv_batch call (unique IDs, first-seen order); excess stay bare. MAX_OSV_VULN_HYDRATIONS = 100 SEARCH_API = "https://search.maven.org/solrsearch/select" # search_artifacts limit: coerced to int and clamped in the handler before the # Solr URL is built (MCP schema bounds are advisory client metadata only). SEARCH_LIMIT_DEFAULT = 10 SEARCH_LIMIT_MAX = 100 # get_dependency_vulnerabilities dependency-list cap — same bound and rationale # as verify_coordinates (enforced in-handler before any network I/O). MAX_VULN_DEPENDENCIES = 100 # get_dependency_license batch cap — same bound/rationale as vuln/verify. MAX_LICENSE_DEPENDENCIES = 100 # Persistent file cache TTLs and capacity (see FileCache below). TTL_POM = 7 * 86400 # 7 days — release POMs are immutable once published TTL_METADATA = 3600 # 1 hour — version lists change, but not constantly TTL_SEARCH = 3600 # 1 hour — Maven Central Solr search index # Negative-cache TTL for a definitive HTTP 404 (#404): a known-absent # coordinate is not re-fetched within this window. Kept far shorter than the # positive TTLs above — an absent artifact can become present the moment it # is published, unlike a resolved metadata/POM document, which is immutable. TTL_NEGATIVE_404 = 300 # 5 minutes CACHE_MAX_ENTRIES = 2000 # Belt-and-suspenders denylist for http_get_cached. Private Maven repo hosts # are NOT listed — this is a blocklist (not an allowlist) so project-declared # repos at any host are still cached via the call-site-discipline path. _CACHE_DENYLIST = frozenset({"api.github.com", "api.osv.dev"}) GRADLE_BUILD_FILES = ["build.gradle.kts", "build.gradle"] GRADLE_SETTINGS_FILES = ["settings.gradle.kts", "settings.gradle"] # #401 collapsed dependency resolution from `1 + P*(1+C)` separate `gradlew` # subprocess calls (each individually budgeted at the old 120s) into ONE # `--init-script` invocation that must resolve every project's configurations # in the SAME window — the timeout's scope changed from "one config on one # module" to "the whole project graph", so the budget is raised accordingly. # Still overridable per-project via MAVEN_MCP_GRADLE_TIMEOUT (see # `_gradle_resolve_timeout_seconds`) for very large multi-module builds. GRADLE_RESOLVE_TIMEOUT = 300 MAX_MODULE_DEPTH = 5 # Cap recursive BOM import / parent-property fetches (#286). MAX_BOM_DEPTH = 5 # deps.dev GetDependencies (#287). Caching is allowed by their ToS; TTL matches # metadata/search (version graphs change, but not constantly). DEPSDEV_API = DEPSDEV_API_DEFAULT TTL_DEPSDEV = 3600 # 1 hour # Fan-out / size caps — Wave 0 hardening: never unbounded network or memory. MAX_TRANSITIVE_GRAPH_NODES = 2000 MAX_CONFLICT_SCAN_ROOTS = 50 MAX_DEPSDEV_ERRORS_REPORTED = 20 # License compliance across transitive trees (#289): GetDependencies for the # graph + GetVersion per unique node for SPDX licenses. Caps bound fan-out. MAX_LICENSE_COMPLIANCE_ROOTS = 20 MAX_LICENSE_COMPLIANCE_NODES = 500 # Upgrade-closure preview. Substitutions are rejected (not truncated) before # any resolve; the diff cap only shortens what is returned; the license cap # bounds GetVersion fan-out to the changed set, not the whole tree. MAX_UPGRADE_SUBSTITUTIONS = 20 MAX_UPGRADE_DIFF_ROWS = 100 MAX_UPGRADE_LICENSE_NODES = 40 # Bounded parallel fan-out (#400): the batch tools below resolve N coordinates # (and, for a few, M repos/roots per coordinate) over the network. Each such # loop is executed via `_map_parallel`, which creates and tears down its OWN # ThreadPoolExecutor per call (never a module-global executor in this # long-lived stdio process). The bound is deliberately modest — some target # hosts (search.maven.org in particular) have a documented rate-limit/403- # lockout history under bulk load (see MAX_GATED_SOLR_CALLS_PER_BATCH below), # so higher concurrency is not free. Calibratable starting point, not proven # against a labeled benchmark. MAX_PARALLEL_FETCHES = 8 # R2c (perf review of #400): handle_get_dependency_health's per-dependency # fan-out makes SEVERAL api.github.com calls each (gh_repo_exists/fetch_repo/ # fetch_releases/issue-stats/discover, plus the rate-limited Search API for # issue stats, plus a deps.dev Scorecard call) — GitHub's SECONDARY rate # limiter specifically penalizes CONCURRENT requests (independent of the # primary per-hour quota, which is 60/h unauthenticated, 5000/h with # GITHUB_TOKEN — see the Environment section), so running this fan-out at the # same width as CDN-metadata/deps.dev/OSV lookups (MAX_PARALLEL_FETCHES) risks # tripping it. A smaller, dedicated bound keeps this ONE fan-out gentler on # GitHub without slowing down the other seven, which have no such penalty. MAX_GITHUB_PARALLEL_FETCHES = 3 # Overall wall-clock budget (#402), in seconds, for a single tool invocation # whose fan-out is otherwise unbounded by any MAX_* cap (audit_project_ # dependencies scans a whole project; check_license_compliance's per-node # GetVersion fetch can run up to MAX_LICENSE_COMPLIANCE_NODES times). When the # deadline is reached mid-batch, the tool returns whatever it already gathered # with `partial: true` instead of blocking indefinitely. Calibratable starting # point — deliberately smaller than MAX_LICENSE_COMPLIANCE_NODES/MAX_PARALLEL_ # FETCHES sequential-worst-case, so a degenerate batch fails fast with partial # data rather than running to real completion. TOOL_DEADLINE = 30.0 # End-of-life / support status via endoflife.date (#415). Support windows are # announced well in advance and change rarely — cache far longer than the # volatile version-list TTLs above. TTL_ENDOFLIFE = 7 * 86400 # 7 days # Gradle configuration names matched by shape (#346), not a closed allow-list: # variant/flavor prefixes (`debugImplementation`, `paidReleaseApi`, …), # source-set forms (`androidTestImplementation`, `testFixturesImplementation`), # annotation processors (`kapt`/`ksp`/`kaptTest`/…), and common standalone # configs (`coreLibraryDesugaring`, `lintChecks`, `detektPlugins`). # Test-ness is classified later via `_is_test_configuration`. _GRADLE_CONFIGURATION_PATTERN = ( r"(?:" # Prefix may be empty (bare `implementation` / `api`) or a variant/flavor/ # source-set name (`debugImplementation`, `androidTestApi`, …). r"\w*(?:[Ii]mplementation|[Cc]ompileOnly|[Rr]untimeOnly|" r"[Aa]nnotationProcessor|[Aa]pi)" r"|kapt(?:AndroidTest|Test)?" r"|ksp(?:AndroidTest|Test)?" r"|coreLibraryDesugaring" r"|lintChecks" r"|detektPlugins" r")" ) SCOPE_TO_CONFIG = { "compile": "implementation", "runtime": "runtimeOnly", "test": "testImplementation", "provided": "compileOnly", "system": "compileOnly", } # --------------------------------------------------------------------------- # HTTP utilities # --------------------------------------------------------------------------- def _make_headers(extra: Optional[Dict[str, str]] = None) -> Dict[str, str]: # Accept-Encoding: gzip (#403) — Maven metadata/POM/search bodies are # 50-200KB text/XML/JSON and compress ~70-80%; _read_response_body # transparently inflates a gzip-encoded response. A caller-supplied # `extra` can still override this, same as any other default header. h = {"User-Agent": USER_AGENT, "Accept-Encoding": "gzip"} if extra: h.update(extra) return h def _github_headers() -> Dict[str, str]: token = os.environ.get("GITHUB_TOKEN") h = _make_headers({"Accept": "application/vnd.github.v3+json"}) if token: h["Authorization"] = f"Bearer {token}" return h # Injectable so tests can monkeypatch it and run instantly (no real sleeping). _sleep: Callable[[float], None] = time.sleep # Injectable clock: monkeypatch _now in tests to control TTL boundary assertions. _now: Callable[[], float] = time.time # Module-level logger targets sys.stderr — the MCP protocol uses stdout for # JSON-RPC; any diagnostic byte written there corrupts the channel. # Guard against double-adding the handler on module re-import (test isolation). _logger = logging.getLogger("maven_mcp") _logger.propagate = False if not _logger.handlers: _logger.addHandler(logging.StreamHandler(sys.stderr)) def _is_retryable_status(status: int) -> bool: """Transient HTTP statuses worth retrying: 429 + any 5xx (matches the TS `isRetriableStatus`). Definitive responses (2xx/3xx/4xx-except-429) are not retried — the caller's tri-state contract treats them as final.""" return status == 429 or 500 <= status < 600 def _backoff_delay(attempt: int) -> float: """Exponential backoff with jitter for retry ``attempt`` (0-based).""" return HTTP_BACKOFF_BASE * (2 ** attempt) + random.random() * HTTP_BACKOFF_JITTER def _parse_retry_after(headers: Any) -> Optional[float]: """Parse a ``Retry-After`` header (delta-seconds or HTTP-date) into seconds, clamped to ``HTTP_RETRY_AFTER_MAX`` so a hostile value cannot stall the server. Returns ``None`` when the header is absent or unparseable.""" if headers is None: return None get = getattr(headers, "get", None) raw = get("Retry-After") if callable(get) else None if not raw: return None raw = raw.strip() try: secs = float(int(raw)) # delta-seconds form except ValueError: try: dt = email.utils.parsedate_to_datetime(raw) # HTTP-date form except (TypeError, ValueError): # 3.9 raises TypeError, 3.10+ raises ValueError on a bad date. return None if dt is None: return None if dt.tzinfo is None: dt = dt.replace(tzinfo=datetime.timezone.utc) secs = (dt - datetime.datetime.now(datetime.timezone.utc)).total_seconds() if secs < 0: secs = 0.0 return min(secs, HTTP_RETRY_AFTER_MAX) # urllib's default opener registers handlers for file:// and ftp://. Repo URLs # are captured verbatim from build files, so a declared ``file:///...`` (or # uppercase ``FILE://``) must never reach urlopen. Only http/https are allowed. _ALLOWED_URL_SCHEMES = frozenset({"http", "https"}) def _url_scheme(url: str) -> str: """Return the URL scheme lowercased, or ``""`` if missing/unparseable.""" try: return urllib.parse.urlsplit(url).scheme.lower() except ValueError: return "" def _assert_http_url(url: str) -> None: """Raise ``URLError`` unless ``url`` uses an allowed http(s) scheme and is not a link-local/cloud-metadata literal IP address. Checked before Request construction so the default opener never honors ``file://`` / ``ftp://`` / other non-HTTP schemes (#348). The link-local check (GHSA-m84v-qqqm-6fr4 follow-up) closes the INITIAL-request half of that SSRF class — a build file can declare ``url = "http://169.254.169.254/…"`` directly, no redirect needed — reusing ``_is_link_local_redirect_target`` (defined later in this module; Python resolves it at call time, not at this function's definition time) so both the initial request and every redirect hop (``_SecureRedirectHandler.redirect_request``, which calls this same function) share one check. RFC1918/loopback stay allowed — private-repo mode legitimately targets internal hosts (#290/#298). Known residuals (not closed here, deliberately not over-engineered): DNS-rebind — a hostname that RESOLVES to a link-local/metadata IP is not caught, since only literal IP hosts are inspected (resolving every host would add a DNS round-trip to every request for a narrow threat); and obfuscated IP encodings (integer/octal/hex forms like ``0xA9FEA9FE`` or ``2852039166``) that ``ipaddress`` rejects as invalid but a libc resolver would still accept — these are not literal dotted-quad/colon-hex strings so ``ipaddress.ip_address()`` raises ``ValueError`` and the check no-ops. """ scheme = _url_scheme(url) if scheme not in _ALLOWED_URL_SCHEMES: raise urllib.error.URLError( f"URL scheme not allowed: {scheme or '(none)'} (only http/https)" ) host = _repo_host(url) if _is_link_local_redirect_target(host): raise urllib.error.URLError( f"Link-local/metadata address blocked: {host}" ) def _is_file_url(url: str) -> bool: """True when ``url`` is a ``file:`` URL (scheme compared case-insensitively).""" return _url_scheme(url) == "file" class ResponseTooLargeError(urllib.error.URLError): """Raised when an HTTP response body exceeds ``HTTP_MAX_RESPONSE_BYTES``. Not retried: re-fetching an oversized body cannot help and would only amplify memory pressure (#350). """ def _response_header(resp: Any, name: str) -> str: """Case-insensitively read a single response header, or ``""`` if absent.""" headers = getattr(resp, "headers", None) if headers is None: return "" get = getattr(headers, "get", None) raw = get(name) if callable(get) else None return (raw or "").strip() def _inflate_gzip_capped(data: bytes, cap: int) -> bytes: """Incrementally inflate a gzip payload, aborting once the DECOMPRESSED size would exceed ``cap`` (#403 zip-bomb guard). A small compressed body can expand enormously (empirically ~750x for a highly repetitive payload) — ``Content-Length`` only ever describes the size on the wire, so it cannot be trusted to bound the inflated size. ``zlib.decompressobj``'s ``max_length`` argument stops production of output at the cap without ever materializing the full inflated buffer, so a hostile stream cannot force a large allocation even transiently; once the cap is exceeded this stops feeding the decompressor further input. """ decompressor = zlib.decompressobj(zlib.MAX_WBITS | 16) # 16+wbits = expect gzip header/trailer try: out = decompressor.decompress(data, cap + 1) while decompressor.unconsumed_tail and len(out) <= cap: out += decompressor.decompress(decompressor.unconsumed_tail, cap + 1 - len(out)) if len(out) <= cap: out += decompressor.flush() except zlib.error as e: # Malformed/corrupt gzip body — treat like any other transport-level # problem so it flows through the existing retry contract instead of # escaping as a novel exception type callers don't expect. raise urllib.error.URLError(f"invalid gzip response body: {e}") from e if len(out) > cap: raise ResponseTooLargeError( f"HTTP response too large: decompressed gzip body exceeds {cap} bytes" ) return out def _read_response_body(resp: Any) -> bytes: """Read ``resp`` body with an explicit size cap (#350). Short-circuits on an oversized ``Content-Length`` before allocating, then reads at most ``HTTP_MAX_RESPONSE_BYTES + 1`` bytes and raises ``ResponseTooLargeError`` if the body exceeds the cap. Chunked / missing Content-Length still cannot grow without bound because of the read cap. A ``Content-Encoding: gzip`` response (sent when the server honors the ``Accept-Encoding: gzip`` request header from ``_make_headers``, #403) is transparently inflated after the raw-bytes cap above has already applied to the WIRE size. The cap is enforced a SECOND time against the DECOMPRESSED size (``_inflate_gzip_capped``) — a small compressed body can still expand far past ``HTTP_MAX_RESPONSE_BYTES`` (zip bomb), and Content-Length only ever describes the wire size, never the inflated one. """ headers = getattr(resp, "headers", None) if headers is not None: get = getattr(headers, "get", None) raw_cl = get("Content-Length") if callable(get) else None if raw_cl is not None and str(raw_cl).strip(): try: content_length = int(str(raw_cl).strip()) except (TypeError, ValueError): content_length = -1 if content_length > HTTP_MAX_RESPONSE_BYTES: raise ResponseTooLargeError( f"HTTP response too large: Content-Length {content_length} " f"exceeds {HTTP_MAX_RESPONSE_BYTES} bytes" ) body = resp.read(HTTP_MAX_RESPONSE_BYTES + 1) if len(body) > HTTP_MAX_RESPONSE_BYTES: raise ResponseTooLargeError( f"HTTP response too large: body exceeds {HTTP_MAX_RESPONSE_BYTES} bytes" ) if _response_header(resp, "Content-Encoding").lower() == "gzip": body = _inflate_gzip_capped(body, HTTP_MAX_RESPONSE_BYTES) return body def _request_with_retry( req: urllib.request.Request, timeout: Optional[float] = None, ) -> Tuple[int, bytes]: """Issue ``req`` with bounded retry/backoff on transient failures. Tri-state contract (relied on by the resolution layer and verify_coordinates): returns ``(status, body)`` for ANY HTTP response — including a persistent 429/5xx after retries are exhausted — and only re-raises the last transport error (URLError / socket.timeout) when EVERY attempt failed at the transport level without ever obtaining an HTTP response. A 4xx (incl. 404) is never turned into a raise. Retry is fully internal and transparent to callers. Oversized bodies raise ``ResponseTooLargeError`` immediately (not retried). ``timeout`` defaults to ``HTTP_TIMEOUT``; enrichment APIs pass ``HTTP_TIMEOUT_EXTERNAL`` (#296). """ if timeout is None: timeout = HTTP_TIMEOUT deadline = time.monotonic() + HTTP_TOTAL_RETRY_BUDGET last_result: Optional[Tuple[int, bytes]] = None last_exc: Optional[BaseException] = None for attempt in range(HTTP_MAX_ATTEMPTS): retry_after: Optional[float] = None try: with _urlopen(req, timeout) as resp: status, body = resp.status, _read_response_body(resp) if not _is_retryable_status(status): return status, body # A retryable status surfaced as a success object (rare — urllib # normally raises HTTPError for 4xx/5xx); remember it and retry. last_result = (status, body) retry_after = _parse_retry_after(getattr(resp, "headers", None)) except ResponseTooLargeError: # Oversized body is definitive — do not retry (#350). raise except urllib.error.HTTPError as e: # HTTPError IS-A URLError but represents a real HTTP response, not a # transport failure — must be caught first. Body stays b"" (the # legacy mapping callers depend on); we never read e.read(). if not _is_retryable_status(e.code): return e.code, b"" last_result = (e.code, b"") retry_after = _parse_retry_after(e.headers) except (urllib.error.URLError, socket.timeout) as e: last_exc = e # http.client.InvalidURL (e.g. a malformed userinfo URL) is deliberately # NOT caught here — it is a request-construction failure, not a transport # failure, so it is not retryable and propagates to the caller. Both # known callers (fetch_metadata and _verify_one) handle it explicitly at # their own layer rather than relying on a generic retry-and-swallow here. if attempt >= HTTP_MAX_ATTEMPTS - 1: break delay = retry_after if retry_after is not None else _backoff_delay(attempt) if time.monotonic() + delay > deadline: break # honoring the delay would blow the total-time budget _sleep(delay) # Prefer any HTTP response seen over a transport error: raise only when no # attempt ever produced an HTTP response. if last_result is not None: return last_result if last_exc is not None: raise last_exc # Defensive: unreachable while HTTP_MAX_ATTEMPTS >= 1 (every attempt sets # last_result or last_exc). Raise an explicit error rather than `raise None`. raise urllib.error.URLError("HTTP request failed: no response and no transport error") def http_get( url: str, headers: Optional[Dict[str, str]] = None, timeout: Optional[float] = None, ) -> Tuple[int, bytes]: """Returns (status_code, body_bytes). Retries transient failures internally (see _request_with_retry); raises urllib.error.URLError / socket.timeout only when every attempt hit a transport error. Non-http(s) schemes are rejected before any network/filesystem open (#348).""" _assert_http_url(url) req = urllib.request.Request(url, headers=headers or _make_headers()) return _request_with_retry(req, timeout=timeout) def http_post_json( url: str, payload: Any, headers: Optional[Dict[str, str]] = None, timeout: Optional[float] = None, ) -> Tuple[int, bytes]: _assert_http_url(url) data = json.dumps(payload).encode() h = _make_headers({"Content-Type": "application/json"}) if headers: h.update(headers) req = urllib.request.Request(url, data=data, headers=h, method="POST") return _request_with_retry(req, timeout=timeout) def http_post_bytes( url: str, data: bytes, content_type: str, headers: Optional[Dict[str, str]] = None, timeout: Optional[float] = None, ) -> Tuple[int, bytes]: """POST raw bytes (e.g. Artifactory AQL ``text/plain`` body).""" _assert_http_url(url) h = _make_headers({"Content-Type": content_type}) if headers: h.update(headers) req = urllib.request.Request(url, data=data, headers=h, method="POST") return _request_with_retry(req, timeout=timeout) # --------------------------------------------------------------------------- # Persistent file cache # --------------------------------------------------------------------------- class FileCache: """On-disk response cache for Maven read paths. Key is sha256(url) hex — a hash is used rather than a path-encoded URL so keys cannot contain path separators regardless of URL content. Headers are intentionally excluded from the key: every wired call site uses the default User-Agent-only headers, and accepting auth-bearing headers not folded into the key would be a latent footgun (cached response served for a different caller's credentials). Security properties: - Only status == 200 or 404 is ever written (#404 negative cache); 429/5xx and other error bodies are never cached. - Each entry JSON stores the full URL; stored url != requested url (hash collision, practically impossible) is treated as a miss. - Cache dir and files use owner-only permissions (0o700 / 0o600). - Disk errors on set() are swallowed and logged to stderr (never stdout, which is the JSON-RPC channel); any error on get() produces a miss. """ def _get_dir(self) -> Optional[str]: """Resolve and ensure the cache directory, per-call (reads os.environ at call time). Returns the path on success, None on OSError (degrade to no-op). Re-running makedirs + chmod on every call is cheap for an already-correct dir and correctly handles the pre-existing loose-perm case (makedirs' mode= is umask-masked and a no-op when the dir pre-exists). """ xdg = os.environ.get("XDG_CACHE_HOME") base = xdg if xdg else os.path.join(os.path.expanduser("~"), ".cache") d = os.path.join(base, "maven-central-mcp") try: os.makedirs(d, mode=0o700, exist_ok=True) # Explicit chmod on the LEAF dir only: makedirs' mode= is # umask-masked and does nothing when the dir pre-exists with # loose perms. The chmod is the reliable tightening path. os.chmod(d, 0o700) return d except OSError: return None def _key_path(self, cache_dir: str, url: str) -> str: h = hashlib.sha256(url.encode("utf-8")).hexdigest() return os.path.join(cache_dir, h + ".json") def get( self, url: str, ttl: float, negative_ttl: Optional[float] = None ) -> Optional[Tuple[int, bytes]]: """Return (status, body) if a fresh entry exists, else None. TTL check is strictly >, so exactly-at-TTL is a HIT. A cached negative (404) entry (#404) is freshness-checked against ``negative_ttl`` instead of ``ttl`` when given, since a definitive-absence result must expire much sooner than a positive metadata/POM/search hit — an absent artifact can become present the moment it is published. Callers that never write 404 entries can omit ``negative_ttl`` and this behaves exactly as before. Any error (missing file, corrupt JSON, OSError, url mismatch) -> None. """ if os.environ.get("MAVEN_MCP_CACHE_DISABLE", "").lower() in ("1", "true", "yes", "on"): return None d = self._get_dir() if d is None: return None path = self._key_path(d, url) try: with open(path, "rb") as fh: entry = json.loads(fh.read()) if entry.get("url") != url: return None status = entry["status"] effective_ttl = ttl if (status == 200 or negative_ttl is None) else negative_ttl if _now() - entry["ts"] > effective_ttl: return None return (status, base64.b64decode(entry["body_b64"])) except Exception: return None def set(self, url: str, status: int, body: bytes) -> None: """Write url->body to cache. No-op for a non-cacheable status, disabled, or dir failure. Cacheable statuses are 200 (positive) and 404 (negative — #404, a clean/definitive absence). Every other status — 429/5xx (transient; `_request_with_retry` already exhausted its own retry budget before returning one of these) and any other 4xx — must always re-hit the network on the next call and is never written. """ if status not in (200, 404): return if os.environ.get("MAVEN_MCP_CACHE_DISABLE", "").lower() in ("1", "true", "yes", "on"): return d = self._get_dir() if d is None: return path = self._key_path(d, url) entry = { "v": 1, "url": url, "status": status, "body_b64": base64.b64encode(body).decode("ascii"), "ts": _now(), } data = json.dumps(entry).encode("utf-8") tmp: Optional[str] = None try: fd, tmp = tempfile.mkstemp(dir=d, suffix=".tmp") try: os.write(fd, data) finally: os.close(fd) os.chmod(tmp, 0o600) os.replace(tmp, path) except OSError as exc: _logger.warning("FileCache set failed for %s: %s", url, exc) if tmp is not None: try: os.unlink(tmp) except OSError: pass # best-effort cleanup of the temp file; a concurrent delete or vanished file is not an error return self._evict(d) def _evict(self, cache_dir: str) -> None: """FIFO eviction: if count > CACHE_MAX_ENTRIES delete oldest-by-mtime down to ~90% of the cap. get() never touches mtime so order is by write time, not access time.""" try: entries = [ os.path.join(cache_dir, f) for f in os.listdir(cache_dir) if f.endswith(".json") ] if len(entries) <= CACHE_MAX_ENTRIES: return entries.sort(key=lambda p: os.stat(p).st_mtime) target = int(CACHE_MAX_ENTRIES * 0.9) for p in entries[: len(entries) - target]: try: os.unlink(p) except (OSError, FileNotFoundError): pass # concurrent delete between listdir and unlink is harmless; skip and continue eviction except OSError: pass # best-effort eviction — a transient I/O error does not affect correctness of the cache _file_cache = FileCache() def _headers_have_authorization(headers: Optional[Dict[str, str]]) -> bool: """True when ``headers`` carries an Authorization value (any casing).""" if not headers: return False return any(str(k).lower() == "authorization" for k in headers) def http_get_cached( url: str, ttl_seconds: float, headers: Optional[Dict[str, str]] = None, timeout: Optional[float] = None, ) -> Tuple[int, bytes]: """Cached GET: returns a cached (200, body) on hit, else delegates to http_get. Auth-bearing requests (``Authorization`` present) NEVER touch the disk cache — the cache key is URL-only, so serving a cached private-repo body to a later caller without the same credentials would be a latent footgun (#291). Public UA-only GETs keep the existing cache path. Sensitive hosts (api.github.com, api.osv.dev) bypass to raw http_get via a static denylist: belt-and-suspenders against future mis-wiring. Private Maven repo hosts are NOT blocked — this is a blocklist, not an allowlist. A cache hit short-circuits ABOVE the #306 retry layer: the response is returned directly without ever calling http_get (and thus without consuming any retry budget). A definitive HTTP 404 is ALSO cached — a negative cache (#404): a known- absent coordinate is not re-fetched on every call within ``TTL_NEGATIVE_404``, which is intentionally far shorter than ``ttl_seconds`` since a 404 can turn into a 200 the moment the artifact is published. 429/5xx and any raised transport error are never cached — matching ``FileCache.set``'s own allow-list — and propagate to the caller exactly as ``http_get`` would return/raise, same as before this change. """ if _headers_have_authorization(headers): return http_get(url, headers, timeout=timeout) host = urllib.parse.urlparse(url).hostname or "" if host in _CACHE_DENYLIST: return http_get(url, headers, timeout=timeout) result = _file_cache.get(url, ttl_seconds, negative_ttl=TTL_NEGATIVE_404) if result is not None: return result status, body = http_get(url, headers, timeout=timeout) if status in (200, 404): _file_cache.set(url, status, body) return (status, body) # --------------------------------------------------------------------------- # Bounded parallel fan-out (#400) # --------------------------------------------------------------------------- def _map_parallel( items: List[Any], fn: Callable[[Any], Any], max_workers: int = MAX_PARALLEL_FETCHES, deadline: Optional[float] = None, ) -> Tuple[List[Any], bool]: """Run ``fn(item)`` for every item in ``items`` on a bounded ThreadPoolExecutor created and torn down within THIS call — never a module-global executor in this long-lived stdio process. Returns ``(results, partial)``: - ``results`` is INDEX-MAPPED to ``items`` (same order as the input), never completion order. Every "first/Nth in order" contract a caller already relies on (``resolvedFrom`` provenance, capability-signal first-wins, error-list ordering, …) is therefore unaffected by which worker thread happens to finish first. - ``partial`` is True only when ``deadline`` was given and was reached before every item finished; unfinished slots are left as ``None`` — the caller decides how to render them (see #402 call sites). PER-ITEM ISOLATION is the caller's responsibility: ``fn`` must catch its own exceptions and return an error-shaped result, exactly mirroring the try/except-and-append-in-order pattern every one of these loops already had before this helper existed. An exception ``fn`` does NOT catch propagates out of ``future.result()`` and thus out of this call — i.e. the whole batch fails, matching what the equivalent sequential loop would have done if it also had no try/except around that step. ``len(items) <= 1`` skips the executor entirely: nothing to parallelize, and it avoids thread-pool startup overhead for the common single-item call (batch tools are very frequently called with exactly one dependency). """ n = len(items) if n == 0: return [], False if n == 1: return [fn(items[0])], False results: List[Any] = [None] * n workers = max(1, min(max_workers, n)) partial = False executor = concurrent.futures.ThreadPoolExecutor(max_workers=workers) future_map: Dict[concurrent.futures.Future, int] = {} try: future_map = {executor.submit(fn, item): i for i, item in enumerate(items)} pending = set(future_map) while pending: timeout: Optional[float] = None if deadline is not None: remaining = deadline - _now() if remaining <= 0: partial = True break timeout = remaining done, pending = concurrent.futures.wait( pending, timeout=timeout, return_when=concurrent.futures.FIRST_COMPLETED ) for future in done: results[future_map[future]] = future.result() if deadline is not None and not done: # wait()'s own timeout elapsed with nothing finished yet. partial = True break finally: # Bounded shutdown (#402): never block THIS call past the deadline # waiting on already-running futures — Python cannot kill a running # thread, so cancel() only drops NOT-YET-STARTED futures from the # queue (no new work begins after the deadline) and `wait=False` lets # this call return promptly; any in-flight worker still finishes # naturally in the background (bounded by its own HTTP timeout) and # the pool's worker threads exit on their own once idle post-shutdown # — never leaked past process lifetime, just not joined by this call. for future in future_map: future.cancel() executor.shutdown(wait=not partial) return results, partial # --------------------------------------------------------------------------- # Private Maven repository credentials (#291) # --------------------------------------------------------------------------- # Credentials are NEVER read from build files. Resolution order (first match # wins): environment variables → ~/.m2/settings.xml → # ~/.gradle/gradle.properties. Secrets must never be logged or echoed into # tool-facing JSON — only Authorization headers on the outbound request. _CRED_ENV_PREFIX = "MAVEN_REPO_" def _sanitize_cred_key(identifier: str) -> str: """Turn a repo id or hostname into an env-var fragment: non-alnum → ``_``, uppercased, trimmed of leading/trailing underscores.""" cleaned = re.sub(r"[^A-Za-z0-9]+", "_", (identifier or "").strip()) return cleaned.strip("_").upper() def _repo_host(url: str) -> str: """Hostname from a repo URL, or ``""`` if unparseable / missing.""" try: return (urllib.parse.urlsplit(url).hostname or "").lower() except ValueError: return "" def _repo_id_candidates(entry: Dict[str, Any]) -> List[str]: """Identifiers worth matching credentials against, in priority order. Maven ```` / Gradle ``name = "…"`` land in ``entry["name"]`` when they differ from the URL. Host is always a fallback so a Gradle ``maven { url }`` with no name still matches ``MAVEN_REPO__…`` env vars. """ candidates: List[str] = [] name = (entry.get("name") or "").strip() url = (entry.get("url") or "").strip() if name and name != url and "://" not in name: candidates.append(name) host = _repo_host(url) if host: candidates.append(host) return candidates def _creds_from_user_secret( username: Optional[str], password: Optional[str], token: Optional[str] ) -> Optional[Dict[str, str]]: """Build a credential dict from optional user/password/token pieces. - token alone → Bearer - username + password → Basic - username + token (no password) → Basic with the token as the password (GitHub Packages / Artifactory PAT style) Incomplete pairs return None — never invent a half-auth header. """ user = (username or "").strip() or None pwd = (password or "").strip() or None tok = (token or "").strip() or None if tok and not user and not pwd: return {"type": "bearer", "token": tok} if user and pwd: return {"type": "basic", "username": user, "password": pwd} if user and tok: return {"type": "basic", "username": user, "password": tok} return None def _resolve_creds_from_env(identifier: str) -> Optional[Dict[str, str]]: """``MAVEN_REPO__USER`` / ``_PASSWORD`` / ``_TOKEN`` (see #291).""" key = _sanitize_cred_key(identifier) if not key: return None prefix = f"{_CRED_ENV_PREFIX}{key}_" return _creds_from_user_secret( os.environ.get(prefix + "USER"), os.environ.get(prefix + "PASSWORD"), os.environ.get(prefix + "TOKEN"), ) def _extract_tag(xml: str, tag: str) -> Optional[str]: r"""First ``content`` leaf value, or None if absent. Shared implementation for the ``re.search(r"([^<]+)", xml)`` idiom repeated throughout this module's regex-only XML parsing: first match, non-empty non-``<`` content, group(1) or None. NOT a fit for a call site needing block-level capture (``[\s\S]*?``), whitespace-trimmed or lazy-dot capture, a dynamic tag name, or empty-tag support (``[^<]*``) — those have different semantics and keep their own inline ``re.search``. """ m = re.search(rf"<{re.escape(tag)}>([^<]+)", xml) return m.group(1) if m else None def _parse_settings_xml_servers(xml: str) -> Dict[str, Dict[str, str]]: """Parse Maven ``settings.xml`` ```` entries into ``{id: {username?, password?}}``. Regex-only (no XML parser dependency). Password values are kept in memory only for header construction — never logged. Malformed / comment-only files yield an empty map.""" servers: Dict[str, Dict[str, str]] = {} # Strip XML comments so a commented-out cannot contribute. xml = re.sub(r"", "", xml, flags=re.DOTALL) for sm in re.finditer(r"([\s\S]*?)", xml): block = sm.group(1) idm = _extract_tag(block, "id") if not idm: continue sid = idm.strip() if not sid: continue user_m = re.search(r"([^<]*)", block) pass_m = re.search(r"([^<]*)", block) entry: Dict[str, str] = {} if user_m: entry["username"] = user_m.group(1) if pass_m: entry["password"] = pass_m.group(1) if entry: servers[sid] = entry return servers def _load_settings_xml_servers() -> Dict[str, Dict[str, str]]: """Read ``~/.m2/settings.xml`` (or ``$M2_HOME/conf/settings.xml`` is NOT consulted — user settings only). Missing/unreadable → empty map.""" path = os.path.join(os.path.expanduser("~"), ".m2", "settings.xml") try: with open(path, "r", encoding="utf-8", errors="ignore") as fh: return _parse_settings_xml_servers(fh.read()) except OSError: return {} def _resolve_creds_from_settings(identifier: str) -> Optional[Dict[str, str]]: """Match a Maven ``settings.xml`` ```` to ``identifier``.""" if not identifier or "://" in identifier: return None servers = _load_settings_xml_servers() entry = servers.get(identifier) if not entry: # Case-insensitive id match — Maven ids are conventionally exact, but # a mismatched case should not silently drop working credentials. lower = identifier.lower() for sid, val in servers.items(): if sid.lower() == lower: entry = val break if not entry: return None return _creds_from_user_secret( entry.get("username"), entry.get("password"), None ) def _parse_gradle_properties(text: str) -> Dict[str, str]: """Minimal ``gradle.properties`` parser: ``key=value`` lines, ``#`` comments, no interpolation. Values keep surrounding spaces stripped.""" props: Dict[str, str] = {} for line in text.splitlines(): raw = line.strip() if not raw or raw.startswith("#") or raw.startswith("!"): continue if "=" not in raw: continue key, _, value = raw.partition("=") key = key.strip() if key: props[key] = value.strip() return props def _load_gradle_properties() -> Dict[str, str]: """Read ``~/.gradle/gradle.properties``. Missing/unreadable → empty map.""" path = os.path.join(os.path.expanduser("~"), ".gradle", "gradle.properties") try: with open(path, "r", encoding="utf-8", errors="ignore") as fh: return _parse_gradle_properties(fh.read()) except OSError: return {} def _resolve_creds_from_gradle_properties(identifier: str) -> Optional[Dict[str, str]]: """Match common Gradle property naming for a repo id: - ``{id}Username`` / ``{id}Password`` (Gradle docs style) - ``{id}User`` / ``{id}Password`` - ``{id}Token`` (Bearer; or Basic password when paired with Username) """ if not identifier or "://" in identifier: return None props = _load_gradle_properties() if not props: return None # Preserve the identifier's own camel/Pascal form for key lookup, and also # try a lowercased-first-letter variant (`nexus` + `Username`). bases = [identifier] if identifier[0].isupper(): bases.append(identifier[0].lower() + identifier[1:]) elif identifier[0].islower(): bases.append(identifier[0].upper() + identifier[1:]) for base in bases: user = props.get(base + "Username") or props.get(base + "User") password = props.get(base + "Password") token = props.get(base + "Token") creds = _creds_from_user_secret(user, password, token) if creds: return creds return None def _name_pin_host(name: str) -> Optional[str]: """``MAVEN_REPO__HOST`` — pins a name/id-keyed credential to one destination host (GHSA-m2hv-xh72-cccw). ``name`` (Maven ```` / Gradle ``name = "…"``) comes from the untrusted, scanned build file, so unlike the hostname (always derived from the same ``url`` a request is actually sent to) it cannot be trusted on its own — a malicious build file could otherwise set ``name`` to any string a *different*, trusted host's secret happens to be keyed under, redirecting that secret to an attacker URL. Returns the pinned hostname (lowercased) or ``None`` when unset. """ key = _sanitize_cred_key(name) if not key: return None raw = os.environ.get(f"{_CRED_ENV_PREFIX}{key}_HOST") return (raw or "").strip().lower() or None def resolve_repo_credentials(entry: Dict[str, Any]) -> Optional[Dict[str, str]]: """Resolve Basic/Bearer credentials for a discovered repo entry (#291). Returns ``{"type": "basic", "username", "password"}`` or ``{"type": "bearer", "token"}``, or ``None`` when nothing matches. Never reads credentials from build files. Never logs secret values. Credential-misbinding guard (GHSA-m2hv-xh72-cccw): a secret must only ever be sent to the host it was configured for. Host-keyed resolution (``ident == host``) is safe unconditionally — the identifier is derived from the SAME ``url`` the request targets, so it cannot diverge from the real destination. Name/id-keyed resolution uses a string the untrusted build file controls independently of ``url``, so it is only honored when the user has pinned that name to this exact destination host via ``MAVEN_REPO__HOST`` (see ``_name_pin_host``); otherwise it is skipped (logged once, secret-free) and resolution falls through to the next candidate (typically the host). Mirror exception (#294 regression fix, R2b): when ``entry["mirrored"]`` is set, ``entry["name"]`` is the settings.xml mirror's OWN ```` and ``entry["url"]`` is the mirror's OWN url — both written by ``_apply_mirror_to_entry`` from ``ctx.mirrors`` (loaded from the user's own ``~/.m2/settings.xml`` / ``~/.gradle/init.gradle*``, never from the scanned project's build files). That name is therefore just as trusted as the host, not an untrusted build-file string, so it does not need a ``_HOST`` pin. This key is set in exactly one place in this module — ``_apply_mirror_to_entry`` — always alongside that url/name overwrite. No parser that reads the scanned project's build files (Gradle/Maven repo parsers, ``discover_repositories``) ever sets or reads this key, so an attacker-controlled build file cannot forge it. """ host = _repo_host(entry.get("url") or "") resolvers = ( _resolve_creds_from_env, _resolve_creds_from_settings, _resolve_creds_from_gradle_properties, ) for ident in _repo_id_candidates(entry): creds = None for resolver in resolvers: creds = resolver(ident) if creds: break if not creds: continue if ident == host or entry.get("mirrored"): return creds pinned_host = _name_pin_host(ident) if pinned_host and pinned_host == host: return creds _logger.warning( "Skipping name/id-keyed credential for %r: set MAVEN_REPO_%s_HOST=%s " "to trust it for this destination (GHSA-m2hv-xh72-cccw)", ident, _sanitize_cred_key(ident), host or "", ) return None def _authorization_header(creds: Dict[str, str]) -> str: """Build an ``Authorization`` header value from a resolve_repo_credentials result.""" if creds.get("type") == "bearer": return f"Bearer {creds['token']}" raw = f"{creds['username']}:{creds['password']}" encoded = base64.b64encode(raw.encode("utf-8")).decode("ascii") return f"Basic {encoded}" def _repo_request_headers(entry: Dict[str, Any]) -> Dict[str, str]: """UA headers plus optional Authorization for ``entry`` (#291).""" creds = resolve_repo_credentials(entry) if not creds: return _make_headers() return _make_headers({"Authorization": _authorization_header(creds)}) def _repo_http_get( entry: Dict[str, Any], url: str, ttl_seconds: Optional[float] = None, ) -> Tuple[int, bytes]: """GET a Maven-repo URL with per-repo auth headers. When ``ttl_seconds`` is set, uses ``http_get_cached`` (auth-bearing responses still bypass the disk cache).""" headers = _repo_request_headers(entry) if ttl_seconds is None: return http_get(url, headers) return http_get_cached(url, ttl_seconds, headers) def _auth_required_message(entry: Dict[str, Any], status: int) -> str: """Clear, secret-free signal when a private repo rejects the request.""" label = _strip_userinfo(entry.get("name") or entry.get("url") or "repository") return f"auth required for {label} (HTTP {status})" # --------------------------------------------------------------------------- # Maven mirrors + closed/offline mode (#294) # --------------------------------------------------------------------------- # Closed-perimeter builds redirect mavenCentral()/google()/gradlePluginPortal() # via settings.xml … (and optionally # MAVEN_MCP_REPOSITORY_BASE / MAVEN_MCP_OFFLINE). Without this, every lookup # hits unreachable public hosts and hangs on timeouts. # Well-known public repo URL → Maven-style repository id(s) used by mirrorOf. _PUBLIC_REPO_MIRROR_IDS = { MAVEN_CENTRAL_URL.rstrip("/"): ("central", "Maven Central"), GOOGLE_MAVEN_URL.rstrip("/"): ("google", "Google Maven"), GRADLE_PLUGIN_PORTAL_URL.rstrip("/"): ("gradle-plugins", "Gradle Plugin Portal"), } def _env_flag(name: str) -> bool: """True when env ``name`` is a truthy toggle (1/true/on/yes).""" return os.environ.get(name, "").strip().lower() in ("1", "true", "on", "yes") def _gradle_resolve_timeout_seconds() -> int: """``MAVEN_MCP_GRADLE_TIMEOUT`` — override for the single-invocation Gradle resolve timeout (#401; see the ``GRADLE_RESOLVE_TIMEOUT`` comment above for why the default was raised). Read fresh on every call (not memoized) so tests can override via ``unittest.mock.patch.dict``. Missing, non-integer, or non-positive values fall back to ``GRADLE_RESOLVE_TIMEOUT`` rather than raising — a malformed override must degrade, not break resolution.""" raw = (os.environ.get("MAVEN_MCP_GRADLE_TIMEOUT") or "").strip() if not raw: return GRADLE_RESOLVE_TIMEOUT try: value = int(raw) except ValueError: return GRADLE_RESOLVE_TIMEOUT return value if value > 0 else GRADLE_RESOLVE_TIMEOUT def _offline_enabled() -> bool: """``MAVEN_MCP_OFFLINE`` — disable all public well-known repo contact (#294).""" return _env_flag("MAVEN_MCP_OFFLINE") # --------------------------------------------------------------------------- # External enrichment services — air-gapped degradation (#296) # --------------------------------------------------------------------------- # OSV / GitHub / deps.dev / developer.android.com / endoflife.date are # unreachable in closed contours. Offline mode short-circuits them (unless an # endpoint override points at an internal mirror). Transport failures mark # capabilityUnavailable="unreachable" with a short timeout so tools do not hang. _ANDROID_DOCS_DEFAULT = "https://developer.android.com" def _env_url_base(name: str, default: str) -> str: """Read an absolute URL base from env, stripping a trailing slash.""" raw = (os.environ.get(name) or "").strip() if not raw: return default.rstrip("/") return raw.rstrip("/") def _url_host(url: str) -> str: try: return (urllib.parse.urlsplit(url).hostname or "").lower() except ValueError: return "" def _github_api_base() -> str: """GitHub API root (``MAVEN_MCP_GITHUB_BASE``; GHE uses ``…/api/v3``).""" return _env_url_base("MAVEN_MCP_GITHUB_BASE", GITHUB_API_DEFAULT) def _osv_api_root() -> str: """OSV API root (``MAVEN_MCP_OSV_BASE``). Accepts either ``https://host`` or ``https://host/v1``; querybatch/vulns paths are derived from this.""" raw = _env_url_base("MAVEN_MCP_OSV_BASE", "https://api.osv.dev") if raw.endswith("/querybatch"): raw = raw[: -len("/querybatch")].rstrip("/") return raw def _osv_querybatch_url() -> str: root = _osv_api_root() if root.endswith("/v1"): return f"{root}/querybatch" return f"{root}/v1/querybatch" def _osv_vuln_url(vuln_id: str) -> str: root = _osv_api_root() quoted = urllib.parse.quote(vuln_id, safe="-._") if root.endswith("/v1"): return f"{root}/vulns/{quoted}" return f"{root}/v1/vulns/{quoted}" def _depsdev_api_base() -> str: """deps.dev v3 API root (``MAVEN_MCP_DEPSDEV_BASE``).""" return _env_url_base("MAVEN_MCP_DEPSDEV_BASE", DEPSDEV_API_DEFAULT) def _android_docs_base() -> str: """developer.android.com root (``MAVEN_MCP_ANDROID_DOCS_BASE``).""" return _env_url_base("MAVEN_MCP_ANDROID_DOCS_BASE", _ANDROID_DOCS_DEFAULT) def _endoflife_api_base() -> str: """endoflife.date v1 API root (``MAVEN_MCP_ENDOFLIFE_BASE``, #415).""" return _env_url_base("MAVEN_MCP_ENDOFLIFE_BASE", _ENDOFLIFE_API_DEFAULT) def _external_override_active(service: str) -> bool: """True when an env override points away from the public default host.""" if service == "osv": return _url_host(_osv_api_root()) != "api.osv.dev" if service == "github": return _url_host(_github_api_base()) != "api.github.com" if service == "depsdev": return _url_host(_depsdev_api_base()) != "api.deps.dev" if service == "android_docs": return _url_host(_android_docs_base()) != "developer.android.com" if service == "endoflife": return _url_host(_endoflife_api_base()) != "endoflife.date" return False def _external_capability(service: str) -> Optional[str]: """Return ``"offline"`` when this enrichment service must be skipped. ``MAVEN_MCP_OFFLINE`` short-circuits public defaults. An endpoint override whose host differs from the public default is treated as an internal mirror and remains callable (still under ``HTTP_TIMEOUT_EXTERNAL``). """ if not _offline_enabled(): return None if _external_override_active(service): return None return "offline" def _with_capability(result: Dict[str, Any], capability: Optional[str]) -> Dict[str, Any]: """Attach ``capabilityUnavailable`` when set; leave result otherwise unchanged.""" if capability: result["capabilityUnavailable"] = capability return result # --------------------------------------------------------------------------- # TLS (internal CA) + HTTP(S) proxy (#298) # --------------------------------------------------------------------------- # Closed contours often terminate TLS with a private CA and route egress via # an HTTP proxy. Defaults stay secure: verification ON, no proxy unless env # says so. Escape hatch MAVEN_MCP_INSECURE_TLS is explicit and warned. # Mutable TLS state (dict, not rebinding module globals) so CodeQL does not # flag the memo/warn flags as unused globals under ``global`` writes. _TLS_STATE: Dict[str, Any] = { "insecure_warned": False, "ssl_context_cache": None, # Optional[Tuple[str, ssl.SSLContext]] } # Guards a first-build race on _TLS_STATE["ssl_context_cache"] under the #400 # ThreadPoolExecutor: without it, N concurrent cache-miss callers (e.g. cold # start + the first parallel batch) would each independently build their own # ssl.SSLContext (redundant CA-bundle loading, not a correctness bug — dict # writes are atomic — but wasted work every time). Double-checked below so # the overwhelmingly common cache-HIT path stays lock-free. _TLS_LOCK = threading.Lock() def _insecure_tls_enabled() -> bool: """``MAVEN_MCP_INSECURE_TLS`` — disable TLS verification (off by default).""" return _env_flag("MAVEN_MCP_INSECURE_TLS") def _ca_cert_files() -> List[str]: """Ordered CA bundle paths to trust in addition to the system store. ``MAVEN_MCP_CA_CERT`` is the primary knob; ``SSL_CERT_FILE`` and ``NODE_EXTRA_CA_CERTS`` are also honored so existing enterprise / Node tooling env can be reused without a second copy of the bundle. """ paths: List[str] = [] for name in ("MAVEN_MCP_CA_CERT", "SSL_CERT_FILE", "NODE_EXTRA_CA_CERTS"): raw = (os.environ.get(name) or "").strip() if not raw: continue expanded = os.path.expanduser(raw) if expanded not in paths: paths.append(expanded) return paths def _warn_insecure_tls_once() -> None: if _TLS_STATE["insecure_warned"]: return _TLS_STATE["insecure_warned"] = True _logger.warning( "MAVEN_MCP_INSECURE_TLS is enabled: TLS certificate verification is " "DISABLED for all HTTP(S) requests. Use only as an explicit escape " "hatch; prefer MAVEN_MCP_CA_CERT with your internal CA bundle." ) def _ssl_config_fingerprint() -> str: parts = ["insecure=%s" % ("1" if _insecure_tls_enabled() else "0")] for p in _ca_cert_files(): try: st = os.stat(p) parts.append("%s:%s:%s" % (p, st.st_mtime_ns, st.st_size)) except OSError: parts.append("%s:missing" % p) return "|".join(parts) def _reset_ssl_context_cache() -> None: """Test helper: drop the memoized SSL context.""" _TLS_STATE["ssl_context_cache"] = None _TLS_STATE["insecure_warned"] = False def _ssl_context() -> "ssl.SSLContext": """SSL context for outbound HTTPS: system CAs + optional internal bundle. Verification stays ON unless ``MAVEN_MCP_INSECURE_TLS`` is explicitly set. Thread-safe (#400): the fast path (cache hit) reads ``_TLS_STATE`` without the lock — a dict read is atomic and the worst case on a stale read is falling through to the locked slow path. The slow (rebuild) path re-checks under ``_TLS_LOCK`` so a build that races another thread's build in flight reuses ITS result instead of doing the work twice. """ fp = _ssl_config_fingerprint() cached = _TLS_STATE["ssl_context_cache"] if cached is not None and cached[0] == fp: return cached[1] with _TLS_LOCK: cached = _TLS_STATE["ssl_context_cache"] if cached is not None and cached[0] == fp: return cached[1] if _insecure_tls_enabled(): _warn_insecure_tls_once() ctx = ssl._create_unverified_context() else: ctx = ssl.create_default_context() for ca_path in _ca_cert_files(): if not os.path.isfile(ca_path): _logger.warning( "CA certificate path not found (%s); ignoring", ca_path ) continue try: ctx.load_verify_locations(cafile=ca_path) except (ssl.SSLError, OSError) as e: _logger.warning( "Failed to load CA bundle %s: %s", ca_path, type(e).__name__ ) _TLS_STATE["ssl_context_cache"] = (fp, ctx) return ctx def _env_proxy_url(scheme: str) -> Optional[str]: """Read proxy URL for ``scheme`` from env (uppercase then lowercase).""" upper = "%s_PROXY" % scheme.upper() lower = "%s_proxy" % scheme.lower() raw = (os.environ.get(upper) or os.environ.get(lower) or "").strip() return raw or None def _explicit_env_proxies() -> Optional[Dict[str, str]]: """Proxies from HTTP(S)_PROXY / ALL_PROXY when any is set; else ``None``. Returning ``None`` leaves urllib's default discovery alone (no custom ``ProxyHandler``). When set, we build an opener that prefers these env vars (predictable in closed contours). """ http = _env_proxy_url("http") https = _env_proxy_url("https") all_proxy = _env_proxy_url("all") if not http and not https and not all_proxy: return None proxies: Dict[str, str] = {} if http or all_proxy: proxies["http"] = http or all_proxy # type: ignore[assignment] if https or http or all_proxy: # Prefer HTTPS_PROXY; fall back to HTTP_PROXY / ALL_PROXY for CONNECT. proxies["https"] = https or http or all_proxy # type: ignore[assignment] return proxies def _no_proxy_list() -> List[str]: raw = (os.environ.get("NO_PROXY") or os.environ.get("no_proxy") or "").strip() if not raw: return [] return [p.strip() for p in raw.split(",") if p.strip()] def _hostname_matches_no_proxy(host: str, pattern: str) -> bool: """Match ``host`` against a single NO_PROXY entry.""" host = (host or "").lower().rstrip(".") pattern = (pattern or "").lower().strip() if not host or not pattern: return False if pattern == "*": return True pat = pattern[1:] if pattern.startswith(".") else pattern if host == pat: return True return host.endswith("." + pat) def _proxy_bypass_host(host: str) -> bool: """True when ``host`` is excluded by ``NO_PROXY`` / ``no_proxy``.""" host = (host or "").lower().rstrip(".") if not host: return False if host in ("localhost", "127.0.0.1", "::1"): return True for entry in _no_proxy_list(): if _hostname_matches_no_proxy(host, entry): return True return False def _urlopen(req: urllib.request.Request, timeout: float): """Open ``req`` with TLS context + optional explicit proxy env (#298). Tests patch ``urllib.request.urlopen``; the no-custom-proxy path calls it directly so existing mocks keep working. When HTTP(S)_PROXY is set and the host is not excluded by NO_PROXY, a dedicated opener applies ``ProxyHandler`` + ``HTTPSHandler(context=…)``. """ url = req.full_url scheme = _url_scheme(url) context = _ssl_context() if scheme == "https" else None proxies = _explicit_env_proxies() host = _url_host(url) if proxies else "" use_proxy = bool(proxies) and bool(host) and not _proxy_bypass_host(host) if use_proxy: handlers: List[Any] = [urllib.request.ProxyHandler(proxies)] if context is not None: handlers.append(urllib.request.HTTPSHandler(context=context)) opener = urllib.request.build_opener(*handlers) return opener.open(req, timeout=timeout) if context is not None: return urllib.request.urlopen(req, timeout=timeout, context=context) return urllib.request.urlopen(req, timeout=timeout) # --------------------------------------------------------------------------- # Secure redirect handling (GHSA-xj4p-wm6r-4q3j / GHSA-m84v-qqqm-6fr4) # --------------------------------------------------------------------------- # urllib's default HTTPRedirectHandler forwards every header except # Content-Length/Content-Type to a redirect target verbatim (including # Authorization, even cross-host or on an https->http downgrade), and only # rejects a redirect scheme outside {http, https, ftp, ''} — it never # inspects the destination host at all. A private repo (or api.github.com # with GITHUB_TOKEN) that 302s can therefore leak credentials to another # host, downgrade to plaintext, or steer a follow-up request at a cloud # metadata endpoint (169.254.169.254) or link-local address. # Proxy-Authorization deliberately excluded: urllib.request.Request stores it # capitalized as "Proxy-authorization" (Request.add_header does key.capitalize()), # so remove_header("Proxy-Authorization") here would never match anyway — and # even if it did, proxy auth is scoped to the PROXY connection, not the target # host, so it is correct for it to persist across a redirect through the same # proxy (R2d cleanup — this key was a no-op, not a security gap). _CREDENTIAL_HEADERS = ("Authorization", "Cookie") def _is_link_local_redirect_target(host: str) -> bool: """True when ``host`` is a literal link-local/metadata IP address. Covers IPv4 169.254.0.0/16 (including the 169.254.169.254 cloud-metadata endpoint) and IPv6 fe80::/10. RFC1918 (10/8, 172.16/12, 192.168/16) and loopback are deliberately NOT blocked — private-repo mode legitimately targets internal hosts (#290/#298). Best-effort: only a literal IP is checked; a redirect ``Location`` that is a hostname is not resolved here (the literal-IP metadata case is the concrete threat this closes). Despite the name, this is not redirect-only: ``_assert_http_url`` also calls it, so the SAME check covers the initial request too (a build file declaring ``url = "http://169.254.169.254/…"`` directly, no redirect needed — GHSA-m84v-qqqm-6fr4 follow-up). Kept under this name rather than renamed, to keep this fix's diff to the call site. """ if not host: return False try: return ipaddress.ip_address(host).is_link_local except ValueError: return False # a hostname, not an IP literal class _SecureRedirectHandler(urllib.request.HTTPRedirectHandler): """Re-validates the redirect target and strips credential headers on every hop (installed globally below — see the assignment after this class). ``redirect_request`` receives the fully-resolved absolute ``newurl`` before a new Request is issued, so raising here blocks the hop entirely instead of only warning after the fact. """ def redirect_request(self, req, fp, code, msg, headers, newurl): # FIX (GHSA-m84v-qqqm-6fr4): only http/https may survive a redirect — # the base handler still allows ftp:// — and link-local/metadata IP # literals are blocked outright. _assert_http_url now covers BOTH the # initial request and every redirect hop (R2c follow-up), so the # link-local check that used to live here as a second, separate call # was removed — this single call raises for either failure mode. _assert_http_url(newurl) new_host = _repo_host(newurl) new_req = super().redirect_request(req, fp, code, msg, headers, newurl) if new_req is None: return None # FIX (GHSA-xj4p-wm6r-4q3j): strip credential headers whenever the # redirect crosses hosts (unconditionally, defense in depth) or # downgrades https -> http (same host or not — cleartext either way). old_host = _repo_host(req.full_url) old_scheme = _url_scheme(req.full_url) new_scheme = _url_scheme(newurl) if old_host != new_host or (old_scheme == "https" and new_scheme == "http"): for header in _CREDENTIAL_HEADERS: new_req.remove_header(header) return new_req # Installed as a module-level swap (not passed to individual build_opener() # calls) so it also protects REAL, un-mocked urllib.request.urlopen() calls: # urlopen() builds its own opener internally via build_opener(), which # resolves HTTPRedirectHandler through urllib.request's module globals at # CALL time (verified against the installed stdlib) — so swapping the name # here is picked up by every opener built anywhere in the process, including # _urlopen()'s own proxy branch above, without changing any call site. Tests # that patch urllib.request.urlopen directly replace the whole function and # never construct a handler at all, so this is inert for the mocked suite. if urllib.request.HTTPRedirectHandler is not _SecureRedirectHandler: urllib.request.HTTPRedirectHandler = _SecureRedirectHandler def _repository_base() -> Optional[str]: """``MAVEN_MCP_REPOSITORY_BASE`` — replace public well-known URLs with this base (trailing slash normalised away for storage; callers re-rstrip).""" raw = (os.environ.get("MAVEN_MCP_REPOSITORY_BASE") or "").strip() return raw.rstrip("/") if raw else None def _parse_settings_xml_mirrors(xml: str) -> List[Dict[str, str]]: """Parse ```` entries into ``[{id, url, mirrorOf}]`` (declaration order). Regex-only. Comment-stripped so a commented-out mirror cannot contribute.""" mirrors: List[Dict[str, str]] = [] xml = re.sub(r"", "", xml, flags=re.DOTALL) for mm in re.finditer(r"([\s\S]*?)", xml): block = mm.group(1) idm = _extract_tag(block, "id") urlm = _extract_tag(block, "url") ofm = _extract_tag(block, "mirrorOf") if not urlm or not ofm: continue url = urlm.strip() mirror_of = ofm.strip() if not url or not mirror_of: continue mid = idm.strip() if idm else url mirrors.append({"id": mid or url, "url": url.rstrip("/"), "mirrorOf": mirror_of}) return mirrors def _settings_xml_paths() -> List[str]: """Candidate settings.xml paths in Maven precedence order for #294: ``MAVEN_MCP_SETTINGS`` (-s override) → ``~/.m2/settings.xml`` → ``$M2_HOME/conf/settings.xml`` / ``$MAVEN_HOME/conf/settings.xml``. First readable file wins (user override replaces global, not merges — MVP).""" paths: List[str] = [] override = (os.environ.get("MAVEN_MCP_SETTINGS") or "").strip() if override: paths.append(os.path.expanduser(override)) return paths paths.append(os.path.join(os.path.expanduser("~"), ".m2", "settings.xml")) for home_var in ("M2_HOME", "MAVEN_HOME"): home = (os.environ.get(home_var) or "").strip() if home: paths.append(os.path.join(home, "conf", "settings.xml")) return paths def _load_settings_xml_mirrors() -> List[Dict[str, str]]: """Load mirrors from the first readable settings.xml (see ``_settings_xml_paths``).""" for path in _settings_xml_paths(): try: with open(path, "r", encoding="utf-8", errors="ignore") as fh: return _parse_settings_xml_mirrors(fh.read()) except OSError: continue return [] def _repo_mirror_ids(entry: Dict[str, Any]) -> List[str]: """Ids that ``mirrorOf`` may match for ``entry``: declared name (when not a URL), plus well-known aliases derived from the URL (``central`` for Maven Central, etc.).""" ids: List[str] = [] name = (entry.get("name") or "").strip() url = (entry.get("url") or "").strip().rstrip("/") if name and "://" not in name: ids.append(name) aliases = _PUBLIC_REPO_MIRROR_IDS.get(url) if aliases: for a in aliases: if a not in ids: ids.append(a) # Host-only fallback so a custom name still matches mirrorOf=hostname. host = _repo_host(url) if host and host not in ids: ids.append(host) return ids def _is_external_repo_url(url: str) -> bool: """Maven ``external:*``: not localhost / loopback and not a ``file:`` URL.""" if _is_file_url(url): return False host = _repo_host(url) if not host: return False if host in ("localhost", "127.0.0.1", "::1") or host.endswith(".localhost"): return False return True def _mirror_of_matches(mirror_of: str, entry: Dict[str, Any]) -> bool: """Maven ``mirrorOf`` pattern matching: comma-separated tokens, ``*``, ``external:*``, ``external:http:*``, explicit ids, and ``!id`` exclusions. A pattern matches when at least one positive token matches and no exclusion token matches (Maven DefaultMirrorSelector semantics, simplified).""" tokens = [t.strip() for t in (mirror_of or "").split(",") if t.strip()] if not tokens: return False repo_ids = {i.lower() for i in _repo_mirror_ids(entry)} url = (entry.get("url") or "").strip() scheme = "" try: scheme = (urllib.parse.urlsplit(url).scheme or "").lower() except ValueError: scheme = "" excluded = False positive_match = False has_positive = False for tok in tokens: if tok.startswith("!"): excl = tok[1:].strip().lower() if excl == "*": excluded = True elif excl in repo_ids: excluded = True continue has_positive = True low = tok.lower() if tok == "*": positive_match = True elif low == "external:*": if _is_external_repo_url(url): positive_match = True elif low == "external:http:*": if scheme == "http" and _is_external_repo_url(url): positive_match = True elif low in repo_ids: positive_match = True if excluded: return False if not has_positive: # Only exclusions → nothing is mirrored (degenerate pattern). return False return positive_match def _select_mirror( entry: Dict[str, Any], mirrors: List[Dict[str, str]] ) -> Optional[Dict[str, str]]: """First settings.xml mirror whose ``mirrorOf`` matches ``entry``, or None.""" for mirror in mirrors: if _mirror_of_matches(mirror.get("mirrorOf", ""), entry): return mirror return None def _apply_mirror_to_entry( entry: Dict[str, Any], mirrors: List[Dict[str, str]] ) -> Dict[str, Any]: """Rewrite ``entry`` URL/name to the matching mirror. Mirror ``id`` becomes ``name`` so #291 credential lookup hits ```` for the mirror. Unmatched entries are returned unchanged (shallow copy).""" out = dict(entry) if not mirrors or _is_file_url(out.get("url") or ""): return out mirror = _select_mirror(out, mirrors) if not mirror: return out out["url"] = mirror["url"] out["name"] = mirror["id"] out["mirrored"] = True return out def _is_well_known_public_url(url: str) -> bool: """True when ``url`` is one of the static public well-known bases.""" return url.rstrip("/") in _PUBLIC_REPO_MIRROR_IDS def _rewrite_public_url(url: str, repository_base: Optional[str]) -> str: """Replace a well-known public URL with ``repository_base`` when set.""" if repository_base and _is_well_known_public_url(url): return repository_base return url def _gradle_init_script_paths() -> List[str]: """``~/.gradle/init.gradle[.kts]`` and ``~/.gradle/init.d/*`` scripts.""" root = os.path.join(os.path.expanduser("~"), ".gradle") paths: List[str] = [] for name in ("init.gradle", "init.gradle.kts"): paths.append(os.path.join(root, name)) init_d = os.path.join(root, "init.d") try: for fn in sorted(os.listdir(init_d)): if fn.endswith((".gradle", ".gradle.kts")): paths.append(os.path.join(init_d, fn)) except OSError: pass # init.d missing or unreadable — no init-script mirrors to load return paths _INIT_MAVEN_URL_RE = re.compile( r"""(?x) (?: # maven\s*\(\s*(?:url\s*=\s*)?["'](https?://[^"']+)["'] maven\s*\(\s*(?:url\s*=\s*)?["'](https?://[^"']+)["'] | # url\s*=\s*(?:uri\s*\(\s*)?["'](https?://[^"']+)["'] \burl\s*=\s*(?:uri\s*\(\s*)?["'](https?://[^"']+)["'] | # Groovy: url\s+["'](https?://[^"']+)["'] \burl\s+["'](https?://[^"']+)["'] ) """ ) def _parse_gradle_init_mirror_urls(text: str) -> List[str]: """Extract http(s) Maven repo URLs from a Gradle init script. Used only when the script also references a well-known shorthand (mavenCentral/google/ gradlePluginPortal) — a heuristic for closed-contour redirect init scripts (#294, "where feasible").""" if not re.search( r"\b(?:mavenCentral|google|gradlePluginPortal)\s*\(", text ): # Also accept scripts that clear/replace repositories without naming # the shorthand — require at least one maven { url } style URL then. if "maven" not in text: return [] urls: List[str] = [] seen = set() for m in _INIT_MAVEN_URL_RE.finditer(text): url = next(g for g in m.groups() if g).rstrip("/") if url not in seen and not _is_well_known_public_url(url): seen.add(url) urls.append(url) return urls def _load_gradle_init_mirrors() -> List[Dict[str, str]]: """Feasible Gradle init-script mirror detection (#294): when an init script declares exactly one non-public maven URL (typical Nexus/Artifactory redirect), treat it as a catch-all ``mirrorOf=*`` mirror. Multi-URL scripts are left alone — too ambiguous for a regex MVP.""" found: List[str] = [] for path in _gradle_init_script_paths(): try: with open(path, "r", encoding="utf-8", errors="ignore") as fh: found.extend(_parse_gradle_init_mirror_urls(fh.read())) except OSError: continue # Dedup preserving order. uniq: List[str] = [] seen = set() for u in found: if u not in seen: seen.add(u) uniq.append(u) if len(uniq) != 1: return [] return [{"id": "gradle-init-mirror", "url": uniq[0], "mirrorOf": "*"}] def _load_mirrors() -> List[Dict[str, str]]: """settings.xml mirrors first; if none, fall back to a single Gradle init catch-all mirror when detectable. settings.xml wins entirely when present so a corporate Maven mirror is not diluted by init-script heuristics.""" mirrors = _load_settings_xml_mirrors() if mirrors: return mirrors return _load_gradle_init_mirrors() # --------------------------------------------------------------------------- # Repository resolution # --------------------------------------------------------------------------- def group_path(group_id: str) -> str: return group_id.replace(".", "/") def _metadata_url(base: str, group_id: str, artifact_id: str) -> str: return f"{base.rstrip('/')}/{group_path(group_id)}/{artifact_id}/maven-metadata.xml" def _pom_url(base: str, group_id: str, artifact_id: str, version: str) -> str: gp = group_path(group_id) return f"{base.rstrip('/')}/{gp}/{artifact_id}/{version}/{artifact_id}-{version}.pom" class ResolutionContext: """Repository-resolution context built ONCE at the handler boundary and threaded down to every resolver. ``scoped_repos`` is the ``discover_repositories`` result and doubles as the per-invocation memo (no separate cache map). Closed-mode fields (#294) — ``offline``, ``repository_base``, ``mirrors`` — and ``public_fallback`` are read from the environment at construction, never sniffed in leaf functions.""" def __init__( self, project_path: str, scoped_repos: Dict[str, List[Dict[str, str]]], public_fallback: bool, offline: bool = False, repository_base: Optional[str] = None, mirrors: Optional[List[Dict[str, str]]] = None, ) -> None: self.project_path = project_path self.scoped_repos = scoped_repos self.public_fallback = public_fallback self.offline = offline self.repository_base = repository_base self.mirrors = list(mirrors) if mirrors else [] def _public_fallback_enabled() -> bool: """MAVEN_MCP_PUBLIC_FALLBACK toggle. Default OFF (closed-mode #294 wants it off); when ON, public repos are appended even for projects that declare their own repositories (escape hatch for implicit/inherited-repo builds).""" return _env_flag("MAVEN_MCP_PUBLIC_FALLBACK") def build_resolution_context(args: Dict) -> "ResolutionContext": """Build a ResolutionContext from a tool-call args dict at the handler boundary. project_path defaults to the current working directory; closed- mode toggles and mirrors are read here once, not in the leaf resolvers.""" project_path = args.get("projectPath") or os.getcwd() return ResolutionContext( project_path, discover_repositories(project_path), _public_fallback_enabled(), offline=_offline_enabled(), repository_base=_repository_base(), mirrors=_load_mirrors(), ) def _public_repos(group_id: str, artifact_id: str) -> List[Tuple[str, str]]: """Static well-known routing (most-specific first): Gradle Plugin Portal for plugin markers, Google Maven for the AndroidX/Google group prefixes, Maven Central always last. Used as the public fallback when the project declares no HTTP-queryable repository for the coordinate's scope.""" repos: List[Tuple[str, str]] = [] if artifact_id.endswith(".gradle.plugin"): repos.append(("Gradle Plugin Portal", GRADLE_PLUGIN_PORTAL_URL)) if any(group_id.startswith(p) for p in GOOGLE_MAVEN_GROUPS): repos.append(("Google Maven", GOOGLE_MAVEN_URL)) repos.append(("Maven Central", MAVEN_CENTRAL_URL)) return repos def _finalize_repo_entries( entries: List[Dict[str, Any]], ctx: "ResolutionContext" ) -> List[Dict[str, Any]]: """Apply ``MAVEN_MCP_REPOSITORY_BASE`` rewrite, settings.xml / init-script mirrors, then drop remaining well-known public URLs when ``offline`` (#294). Dedupes by URL (first-seen wins) so a catch-all mirror collapsing several public shorthands does not probe the same host repeatedly.""" out: List[Dict[str, Any]] = [] for entry in entries: e = dict(entry) e["url"] = _rewrite_public_url(e["url"], ctx.repository_base) if ctx.repository_base and _is_well_known_public_url(entry["url"]): # Name becomes the base host so #291 cred lookup can match it. host = _repo_host(ctx.repository_base) or ctx.repository_base e["name"] = host e = _apply_mirror_to_entry(e, ctx.mirrors) if ctx.offline and _is_well_known_public_url(e["url"]): continue out.append(e) return _dedup_repos(out) def _repos_for( group_id: str, artifact_id: str, ctx: "ResolutionContext" ) -> List[Dict[str, Any]]: """Project-first repository resolution. ``ctx`` is REQUIRED — a public-only default would silently resurrect #310. Returns rich entries ``{name, url, scope, is_public_fallback}``, most-specific first. Coordinate kind: a ``.gradle.plugin`` marker artifact resolves in the plugin scope, everything else in the dependency scope. If that scope declares >=1 HTTP-queryable repository (mavenLocal's ``file://`` marker does NOT count) the declared repos are returned EXACTLY, with no implicit public append — the #299/#310 core. Otherwise the static public routing is the fallback. When ``ctx.public_fallback`` is ON the public repos are appended even for a declared scope (deduped by URL). Closed mode (#294): ``ctx.mirrors`` rewrite matched repo URLs (settings.xml ``mirrorOf`` / Gradle init heuristic); ``ctx.repository_base`` replaces well-known public URLs; ``ctx.offline`` drops any remaining public hosts so they are never contacted.""" scope = "plugin" if artifact_id.endswith(".gradle.plugin") else "dependency" declared = ctx.scoped_repos.get(scope, []) # mavenLocal / file:// markers are non-queryable; scheme check is # case-insensitive so ``FILE://`` cannot bypass the guard (#348). queryable = [r for r in declared if not _is_file_url(r["url"])] # When offline with no repository_base and no mirrors, public fallbacks are # suppressed entirely — contacting repo1.maven.org would only hang. public_entries: List[Dict[str, Any]] = [] if (not ctx.offline) or ctx.repository_base or ctx.mirrors: public_entries = [ {"name": name, "url": url, "scope": scope, "is_public_fallback": True} for name, url in _public_repos(group_id, artifact_id) ] if not queryable: return _finalize_repo_entries(public_entries, ctx) # Content/group filtering (#320): a repo with an attached `includeGroup` / # `includeGroupByRegex` filter is only consulted for a coordinate whose # groupId actually matches — mirrors Gradle's own content filtering so a # group-scoped repo (JitPack scoped to `com.github.*`, a filtered Google # Maven declaration, etc.) is never queried for a group the real build # would never route to it (the #299 false-availability class). Unfiltered # repos are unaffected — the filter is opt-in per repo, not a new default # restriction. If filtering excludes every declared repo, the scope is # still treated as "declared" (not empty) — no unconditional public # fallback, only the same opt-in `ctx.public_fallback` append as below. matched = [r for r in queryable if _repo_matches_group(r, group_id)] entries = [ {"name": r["name"], "url": r["url"], "scope": scope, "is_public_fallback": False} for r in matched ] if ctx.public_fallback and public_entries: entries.extend(public_entries) return _finalize_repo_entries(_dedup_repos(entries), ctx) return _finalize_repo_entries(entries, ctx) def _parse_metadata_xml(xml: str, group_id: str, artifact_id: str) -> Dict[str, Any]: versions = re.findall(r"([^<]+)", xml) latest = _extract_tag(xml, "latest") release = _extract_tag(xml, "release") last_updated = _extract_tag(xml, "lastUpdated") return { "groupId": group_id, "artifactId": artifact_id, "versions": versions, "latest": latest, "release": release, "lastUpdated": last_updated, } def _strip_userinfo(url: str) -> str: """Redact ``user:pass@`` userinfo from a repo URL before it reaches tool-facing JSON (#317 security review). Repo URLs are captured verbatim from build files, so a discouraged hardcoded ``url = "https://user:pass@host/repo"`` would otherwise echo the literal credential into MCP output. Output-boundary only — the raw, credentialed URL is still what actually gets HTTP-fetched; this never touches the fetch path, only what is reported back.""" try: parsed = urllib.parse.urlsplit(url) except ValueError: # A malformed host (e.g. an unterminated bracketed IPv6 literal) can # still carry userinfo that urlsplit never gets to parse out — fail # open would leak it verbatim. A first-`@`-only regex (the previous # approach) under-redacts when the userinfo itself contains an # unescaped `@` (e.g. `user:pa@ss@host` — malformed input exactly # like what triggers this fallback): it would strip only up to the # FIRST `@`, leaving a trailing password fragment (`ss@host`) in the # output. Instead, locate the `scheme://` prefix, take the authority # up to the next `/` (or end of string), and drop everything up to # and including the LAST `@` in that authority — that reliably # removes all userinfo regardless of how many `@` it contains. scheme_match = re.match(r"^([a-zA-Z][a-zA-Z0-9+.-]*://)", url) if not scheme_match: # No identifiable scheme boundary — fall back to stripping # everything up to the last `@` globally as a safety net. return url.rsplit("@", 1)[-1] if "@" in url else url scheme = scheme_match.group(1) rest = url[len(scheme):] slash_idx = rest.find("/") authority = rest if slash_idx == -1 else rest[:slash_idx] remainder = "" if slash_idx == -1 else rest[slash_idx:] if "@" not in authority: return url return f"{scheme}{authority.rsplit('@', 1)[-1]}{remainder}" if "@" not in parsed.netloc: return url host = parsed.netloc.rsplit("@", 1)[-1] return urllib.parse.urlunsplit((parsed.scheme, f"***@{host}", parsed.path, parsed.query, parsed.fragment)) def _to_resolved_from(entry: Dict[str, Any]) -> Dict[str, Any]: """Project a repo entry (from _repos_for/_public_repos) into the public resolvedFrom shape.""" return { "url": _strip_userinfo(entry["url"]), "scope": entry["scope"], "viaPublicFallback": entry["is_public_fallback"], } def fetch_metadata(group_id: str, artifact_id: str, ctx: "ResolutionContext") -> Dict[str, Any]: """Query every repo in ``ctx`` and MERGE results across those answering 200: version sets are unioned, deduped and sorted ascending so a private repo's extra versions are not lost to a first-hit short-circuit (#311); ``lastUpdated`` carries the most-recent value across answering repos. If NO repo answers (all 404/error) a ``ValueError`` is raised with the same message as the legacy first-hit code, so unwrapped callers keep working. The single-repo result is identical to the legacy path (union/sort of one ascending set = itself). Intentionally diverges from the retired TS ``resolveAll``: no proxy-dedup.""" repos = _repos_for(group_id, artifact_id, ctx) merged_versions: List[str] = [] last_updated: Optional[str] = None answered = False # An empty `repos` list has no entry to set a last_err from — default to an # explicit reason instead of leaving this None (which would surface as the # confusing "...: None"). Two causes since #320/#294: content/group filtering # excluded every declared repo, or offline/closed mode dropped all public # hosts with no mirror / REPOSITORY_BASE replacement. if not repos: if ctx.offline and not ctx.repository_base and not ctx.mirrors: last_err = ( "no queryable repositories (offline/closed mode with no mirror " "or MAVEN_MCP_REPOSITORY_BASE)" ) else: last_err = ( "no repository in scope (declared repo(s) excluded by " "content/group filtering)" ) else: last_err = None # First repo (in _repos_for order: declared repos before any public-fallback # append) that answers 200 — surfaced as resolvedFrom for #317 provenance. resolved_from: Optional[Dict[str, Any]] = None for entry in repos: url = _metadata_url(entry["url"], group_id, artifact_id) try: status, body = _repo_http_get(entry, url, TTL_METADATA) if status == 200: # "answered" is gated on HTTP 200 itself, not on a non-empty # version list — a 200 with empty still counts as a # reachable repo, matching the legacy first-hit contract. answered = True if resolved_from is None: resolved_from = _to_resolved_from(entry) parsed = _parse_metadata_xml(body.decode("utf-8", errors="replace"), group_id, artifact_id) merged_versions.extend(parsed["versions"]) lu = parsed.get("lastUpdated") if lu and (last_updated is None or lu > last_updated): last_updated = lu elif status in (401, 403): # Private/corporate repos reject unauthenticated (or wrong-cred) # probes with 401/403 — surface a clear "auth required" signal # rather than a bare HTTP code (#291). Never include secrets. last_err = _auth_required_message(entry, status) else: # entry["name"] can be the literal repo URL for maven("url") # declarations (name == url); redact before it reaches the # exception message, which flows into tool-facing "error" fields. last_err = f"HTTP {status} from {_strip_userinfo(entry['name'])}" except Exception as e: # Never interpolate str(e) here: a userinfo URL (https://user:pass@host) # makes urlopen raise http.client.InvalidURL — NOT a urllib.error.URLError, # so it lands in this generic branch — whose message embeds the raw # password (e.g. "nonnumeric port: 'pass@host'"), a string shape # _strip_userinfo cannot redact (it only redacts a value that IS a bare # scheme://user:pass@host URL, not a credential fragment embedded in # free text). Build the message from known-safe components instead: # the exception type name plus the already-redacted repo name. last_err = f"{type(e).__name__} from {_strip_userinfo(entry['name'])}" if not answered: raise ValueError(f"Could not fetch metadata for {group_id}:{artifact_id}: {last_err}") versions = sorted(set(merged_versions), key=functools.cmp_to_key(compare_versions)) return { "groupId": group_id, "artifactId": artifact_id, "versions": versions, "latest": find_latest_version(versions, "PREFER_STABLE"), "release": find_latest_version(versions, "STABLE_ONLY"), "lastUpdated": last_updated, "resolvedFrom": resolved_from, } def check_version_in_repos(group_id: str, artifact_id: str, version: str, ctx: "ResolutionContext") -> Optional[Dict[str, Any]]: """Returns the matching repo entry (name/url/scope/is_public_fallback) if version exists, else None.""" repos = _repos_for(group_id, artifact_id, ctx) for entry in repos: url = _metadata_url(entry["url"], group_id, artifact_id) try: status, body = _repo_http_get(entry, url, TTL_METADATA) if status == 200: xml = body.decode("utf-8", errors="replace") versions = re.findall(r"([^<]+)", xml) if version in versions: return entry except Exception: continue return None def fetch_pom(group_id: str, artifact_id: str, version: str, ctx: "ResolutionContext") -> Optional[str]: repos = _repos_for(group_id, artifact_id, ctx) for entry in repos: url = _pom_url(entry["url"], group_id, artifact_id, version) try: status, body = _repo_http_get(entry, url, TTL_POM) if status == 200: return body.decode("utf-8", errors="replace") except Exception: continue return None def check_relocation( group_id: str, artifact_id: str, version: str, ctx: "ResolutionContext" ) -> Optional[Dict[str, str]]: """Fetch a resolved coordinate's POM and report Maven relocation, if any (#284). Returns None when the POM can't be fetched (repo miss, transport failure — degrades silently, same as any other optional POM-derived field) or carries no `` block. Reuses `fetch_pom`'s cache (POM TTL 7 days), so this is one extra cached network call on top of whatever the caller already made to resolve `version` — not a new resolution pass.""" pom = fetch_pom(group_id, artifact_id, version, ctx) if pom is None: return None return extract_relocation_from_pom(pom, group_id, artifact_id, version) # --------------------------------------------------------------------------- # BOM / platform expansion (#286) # --------------------------------------------------------------------------- def _parse_pom_properties(pom_xml: str) -> Dict[str, str]: """Parse `` children via regex. Returns {name: value}.""" xml = _strip_xml_comments(pom_xml) props_m = re.search(r"([\s\S]*?)", xml) if not props_m: return {} props: Dict[str, str] = {} for m in re.finditer(r"<([a-zA-Z0-9_.-]+)>([^<]*)", props_m.group(1)): props[m.group(1)] = m.group(2).strip() return props def _pom_project_property_defaults( pom_xml: str, bom_version: Optional[str] = None ) -> Dict[str, str]: """Build project.* / bare groupId/artifactId/version property defaults from a POM's own coordinates (with parent fallback for groupId/version).""" coords = _parse_maven_project_coords(pom_xml) parent = _parse_maven_parent(pom_xml) gid = coords.get("groupId") or (parent.get("groupId") if parent else None) aid = coords.get("artifactId") ver = ( coords.get("version") or bom_version or (parent.get("version") if parent else None) ) defaults: Dict[str, str] = {} if gid: defaults["project.groupId"] = gid defaults["groupId"] = gid if aid: defaults["project.artifactId"] = aid defaults["artifactId"] = aid if ver: defaults["project.version"] = ver defaults["version"] = ver return defaults def _interpolate_pom_props(value: Optional[str], props: Dict[str, str]) -> Optional[str]: """Multi-pass `${name}` substitution. Returns None when value is None. Leaves unresolved `${...}` fragments intact for the caller to skip.""" if value is None: return None if "${" not in value: return value result = value for _ in range(10): if "${" not in result: break def _repl(m: re.Match) -> str: return props.get(m.group(1), m.group(0)) new = re.sub(r"\$\{([^}]+)\}", _repl, result) if new == result: break result = new return result def _collect_bom_properties( pom_xml: str, bom_version: str, ctx: "ResolutionContext", depth: int = 0, ) -> Dict[str, str]: """Merge parent POM properties under child (child wins). Parent is fetched by GAV via `fetch_pom` — remote BOMs have no local relativePath.""" props: Dict[str, str] = {} parent = _parse_maven_parent(pom_xml) parent_group_id = parent.get("groupId") if parent else None parent_artifact_id = parent.get("artifactId") if parent else None parent_version = parent.get("version") if parent else None if ( parent_group_id and parent_artifact_id and parent_version and depth < MAX_BOM_DEPTH ): parent_pom = fetch_pom( parent_group_id, parent_artifact_id, parent_version, ctx ) if parent_pom: props.update( _collect_bom_properties( parent_pom, parent_version, ctx, depth + 1 ) ) props.update(_pom_project_property_defaults(pom_xml, bom_version)) props.update(_parse_pom_properties(pom_xml)) return props def parse_dependency_management( pom_xml: str, props: Optional[Dict[str, str]] = None ) -> List[Dict]: """Parse `…` entries (#286). Each entry: groupId, artifactId, version, scope (default compile), type (default jar), isImportBom (scope==import and type==pom). Strips XML comments first. When ``props`` is omitted, interpolates using this POM's own `` plus project.* defaults (no network). """ xml = _strip_xml_comments(pom_xml) dm_m = re.search( r"([\s\S]*?)", xml ) if not dm_m: return [] deps_m = re.search(r"([\s\S]*?)", dm_m.group(1)) block = deps_m.group(1) if deps_m else dm_m.group(1) if props is None: props = {} props.update(_pom_project_property_defaults(pom_xml)) props.update(_parse_pom_properties(pom_xml)) results: List[Dict] = [] for m in re.finditer(r"([\s\S]*?)", block): dep_block = m.group(1) gid_m = _extract_tag(dep_block, "groupId") aid_m = _extract_tag(dep_block, "artifactId") if not gid_m or not aid_m: continue group_id = _interpolate_pom_props(gid_m.strip(), props) artifact_id = _interpolate_pom_props(aid_m.strip(), props) if not group_id or not artifact_id: continue if "${" in group_id or "${" in artifact_id: continue ver_m = _extract_tag(dep_block, "version") version = ( _interpolate_pom_props(ver_m.strip(), props) if ver_m else None ) if version is not None and "${" in version: version = None scope_m = _extract_tag(dep_block, "scope") scope = scope_m.strip() if scope_m else "compile" type_m = _extract_tag(dep_block, "type") dep_type = type_m.strip() if type_m else "jar" results.append({ "groupId": group_id, "artifactId": artifact_id, "version": version, "scope": scope, "type": dep_type, "isImportBom": scope == "import" and dep_type == "pom", }) return results def expand_bom( group_id: str, artifact_id: str, version: str, ctx: "ResolutionContext", *, _depth: int = 0, _seen: Optional[set] = None, ) -> List[Dict]: """Fetch a BOM POM and return managed ``[{groupId, artifactId, version}]``. Import-scope BOMs are expanded recursively and merged with Maven's first-wins ordering (document order; do not overwrite an existing GA). Direct managed entries in the current BOM follow the same first-wins rule. Depth-capped and cycle-guarded on ``g:a:v``. Missing POM / unresolved ``${}`` versions are skipped — never raises (#286). """ if _seen is None: _seen = set() key = f"{group_id}:{artifact_id}:{version}" if _depth > MAX_BOM_DEPTH or key in _seen: return [] _seen.add(key) pom = fetch_pom(group_id, artifact_id, version, ctx) if not pom: return [] props = _collect_bom_properties(pom, version, ctx, _depth) entries = parse_dependency_management(pom, props=props) managed: List[Dict] = [] managed_keys: set = set() for entry in entries: if entry["isImportBom"]: if not entry.get("version"): continue nested = expand_bom( entry["groupId"], entry["artifactId"], entry["version"], ctx, _depth=_depth + 1, _seen=_seen, ) for item in nested: ga = (item["groupId"], item["artifactId"]) if ga in managed_keys: continue managed_keys.add(ga) managed.append(item) continue if not entry.get("version"): continue ga = (entry["groupId"], entry["artifactId"]) if ga in managed_keys: continue managed_keys.add(ga) managed.append({ "groupId": entry["groupId"], "artifactId": entry["artifactId"], "version": entry["version"], }) return managed def apply_bom_managed_versions(scan: Dict, ctx: "ResolutionContext") -> Dict: """Apply BOM/platform managed versions onto scanned dependencies (#286). Expands each ``isPlatform`` dependency via ``expand_bom`` in declaration order (first-wins), then lets local Maven non-import ``dependencyManagement`` pins override imported managed versions. Versionless non-platform deps gain ``effectiveVersion`` + ``managedBy``. Explicit versions win unless the contributing platform was ``enforcedPlatform``. """ managed_map: Dict[Tuple[str, str], Dict] = {} for dep in scan.get("dependencies") or []: if not dep.get("isPlatform"): continue ver = dep.get("version") gid, aid = dep.get("groupId"), dep.get("artifactId") if not gid or not aid or not ver: continue bom_ref = {"groupId": gid, "artifactId": aid, "version": ver} kind = dep.get("platformKind") or "platform" for item in expand_bom(gid, aid, ver, ctx): ga = (item["groupId"], item["artifactId"]) if ga in managed_map: continue managed_map[ga] = { "version": item["version"], "managedBy": dict(bom_ref), "platformKind": kind, } # Local non-import pins override imported managed versions (Maven direct # declaration override). First local pin for a GA wins among local pins. local_applied: set = set() for pin in scan.get("managedPins") or []: gid, aid, ver = pin.get("groupId"), pin.get("artifactId"), pin.get("version") if not gid or not aid or not ver: continue ga = (gid, aid) if ga in local_applied: continue local_applied.add(ga) managed_map[ga] = { "version": ver, "managedBy": {"groupId": gid, "artifactId": aid, "version": ver}, "platformKind": None, } for dep in scan.get("dependencies") or []: if dep.get("isPlatform"): continue gid, aid = dep.get("groupId"), dep.get("artifactId") if not gid or not aid: continue info = managed_map.get((gid, aid)) if not info: continue explicit = dep.get("version") if explicit: if ( info.get("platformKind") == "enforcedPlatform" and explicit != info["version"] ): dep["effectiveVersion"] = info["version"] dep["managedBy"] = info["managedBy"] continue dep["effectiveVersion"] = info["version"] dep["managedBy"] = info["managedBy"] return scan # --------------------------------------------------------------------------- # deps.dev transitive graphs + conflict detection (#287) # --------------------------------------------------------------------------- def _depsdev_package_name(group_id: str, artifact_id: str) -> str: """Maven package name on deps.dev is ``groupId:artifactId``.""" return f"{group_id}:{artifact_id}" def _split_maven_package_name(name: str) -> Optional[Tuple[str, str]]: """Split a deps.dev Maven ``name`` (``g:a``) into groupId/artifactId. Returns ``None`` when the name is not a two-part Maven coordinate (e.g. bundled npm-style encodings). Callers skip those nodes. """ if not name or ":" not in name: return None parts = name.split(":") if len(parts) != 2 or not parts[0] or not parts[1]: return None return parts[0], parts[1] def _depsdev_dependencies_url(group_id: str, artifact_id: str, version: str) -> str: """Build the GetDependencies URL. Path segments are percent-encoded.""" name = _depsdev_package_name(group_id, artifact_id) base = _depsdev_api_base() return ( f"{base}/systems/MAVEN/packages/" f"{urllib.parse.quote(name, safe='')}/versions/" f"{urllib.parse.quote(version, safe='')}:dependencies" ) def _depsdev_version_url(group_id: str, artifact_id: str, version: str) -> str: """Build the GetVersion URL (licenses live here, not on GetDependencies).""" name = _depsdev_package_name(group_id, artifact_id) base = _depsdev_api_base() return ( f"{base}/systems/MAVEN/packages/" f"{urllib.parse.quote(name, safe='')}/versions/" f"{urllib.parse.quote(version, safe='')}" ) def _depsdev_project_url(owner: str, repo: str) -> str: """Build the deps.dev v3 GetProject URL for a GitHub repository (#411). ``projectKey`` embeds literal ``/`` separators (``github.com/{owner}/{repo}``) and is percent-encoded as a SINGLE path segment — verified live against api.deps.dev: mixed-case owner/repo is accepted and normalized server-side, but an uppercase host segment is rejected with HTTP 400 "invalid project key", which is why the ``github.com`` literal here is always lowercase. """ key = f"github.com/{owner}/{repo}" base = _depsdev_api_base() return f"{base}/projects/{urllib.parse.quote(key, safe='')}" def fetch_depsdev_licenses( group_id: str, artifact_id: str, version: str, ) -> Dict[str, Any]: """Fetch SPDX license strings for one GAV via deps.dev GetVersion (#289). Returns ``{ok, status, licenses, error}``. Never raises for network/HTTP/ parse failures — callers degrade to an empty license list + error. ``licenses`` are the raw deps.dev strings (SPDX expressions or ``non-standard``); empty means deps.dev had no license metadata. """ empty: Dict[str, Any] = { "ok": False, "status": None, "licenses": [], "error": None, } cap = _external_capability("depsdev") if cap: out = dict(empty) out["error"] = "deps.dev unavailable (offline/closed mode)" out["capabilityUnavailable"] = cap return out url = _depsdev_version_url(group_id, artifact_id, version) try: status, body = http_get_cached( url, TTL_DEPSDEV, timeout=HTTP_TIMEOUT_EXTERNAL ) except Exception as e: out = dict(empty) out["error"] = f"{type(e).__name__}: deps.dev unreachable" out["capabilityUnavailable"] = "unreachable" return out out = dict(empty) out["status"] = status if status != 200 or not body: out["error"] = ( f"deps.dev returned HTTP {status}" if status is not None else "deps.dev returned empty response" ) return out try: payload = json.loads(body.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as e: out["error"] = f"deps.dev response not JSON: {type(e).__name__}" return out raw = payload.get("licenses") or [] if not isinstance(raw, list): raw = [str(raw)] licenses = [str(x).strip() for x in raw if str(x).strip()] out.update({"ok": True, "licenses": licenses, "error": None}) return out def fetch_depsdev_scorecard(owner: str, repo: str) -> Dict[str, Any]: """Fetch the OpenSSF Scorecard for a GitHub repo via deps.dev GetProject (#411). Returns ``{ok, status, scorecard, error}``. ``scorecard`` is the trimmed ``{overallScore, date, checks, generatedBy}`` shape when deps.dev has one on file; ``None`` when the deps.dev call itself succeeded but no scorecard is available. deps.dev's project index does not cover every GitHub repository — verified live against api.deps.dev: a 404 plain-text ``"project not found"`` body is the common response for a repo deps.dev has not indexed (e.g. small/personal repos), not a transport failure, so it degrades to ``scorecard: None`` rather than an error callers should alarm on. Never raises for network/HTTP/parse failures — mirrors fetch_depsdev_licenses / fetch_depsdev_dependencies: callers degrade to no scorecard plus an error/capabilityUnavailable. """ empty: Dict[str, Any] = {"ok": False, "status": None, "scorecard": None, "error": None} cap = _external_capability("depsdev") if cap: out = dict(empty) out["error"] = "deps.dev unavailable (offline/closed mode)" out["capabilityUnavailable"] = cap return out url = _depsdev_project_url(owner, repo) try: status, body = http_get_cached( url, TTL_DEPSDEV, timeout=HTTP_TIMEOUT_EXTERNAL ) except Exception as e: out = dict(empty) out["error"] = f"{type(e).__name__}: deps.dev unreachable" out["capabilityUnavailable"] = "unreachable" return out out = dict(empty) out["status"] = status if status != 200 or not body: # Covers the empirically-confirmed 404 "project not found" case (a repo # deps.dev has not indexed) alongside genuine transport/HTTP failures — # both simply mean "no scorecard data", never a crash. out["error"] = ( f"deps.dev returned HTTP {status}" if status is not None else "deps.dev returned empty response" ) return out try: payload = json.loads(body.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as e: out["error"] = f"deps.dev response not JSON: {type(e).__name__}" return out out["ok"] = True raw_sc = payload.get("scorecard") if not isinstance(raw_sc, dict): return out # project known to deps.dev, but no scorecard on file checks: List[Dict[str, Any]] = [] for c in (raw_sc.get("checks") or []): if isinstance(c, dict) and c.get("name"): checks.append({ "name": c.get("name"), "score": c.get("score"), "reason": c.get("reason"), }) out["scorecard"] = { "overallScore": raw_sc.get("overallScore"), "date": raw_sc.get("date"), "checks": checks, "generatedBy": "OpenSSF", } return out def fetch_depsdev_dependencies( group_id: str, artifact_id: str, version: str, ) -> Dict[str, Any]: """Fetch a resolved Maven dependency graph from deps.dev GetDependencies. Returns a normalised dict:: { "ok": bool, "status": int | None, # HTTP status when a response arrived "error": str | None, # human-readable degrade reason "graphError": str | None, # deps.dev graph-level error field "nodes": [{groupId, artifactId, version, relation, errors}], "edges": [{from, to, requirement}], "partial": bool, # True when truncated or degraded "truncated": bool, # True when node/edge cap applied } Never raises for network/HTTP/parse failures — callers get ``ok=False`` with an ``error`` string (graceful degradation AC). Oversized bodies still raise via ``http_get`` / ``ResponseTooLargeError`` only if the transport layer itself raises after retries; those are caught and mapped to ``ok=False``. """ empty: Dict[str, Any] = { "ok": False, "status": None, "error": None, "graphError": None, "nodes": [], "edges": [], "partial": True, "truncated": False, } cap = _external_capability("depsdev") if cap: out = dict(empty) out["error"] = "deps.dev unavailable (offline/closed mode)" out["capabilityUnavailable"] = cap return out url = _depsdev_dependencies_url(group_id, artifact_id, version) try: status, body = http_get_cached( url, TTL_DEPSDEV, timeout=HTTP_TIMEOUT_EXTERNAL ) except Exception as e: # Transport / size-cap / scheme failures — degrade, never raise. out = dict(empty) out["error"] = f"{type(e).__name__}: deps.dev unreachable" out["capabilityUnavailable"] = "unreachable" return out out = dict(empty) out["status"] = status if status != 200 or not body: out["error"] = ( f"deps.dev returned HTTP {status}" if status is not None else "deps.dev returned empty response" ) return out try: payload = json.loads(body.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as e: out["error"] = f"deps.dev response not JSON: {type(e).__name__}" return out raw_nodes = payload.get("nodes") or [] raw_edges = payload.get("edges") or [] graph_error = payload.get("error") or None if isinstance(graph_error, str) and not graph_error.strip(): graph_error = None # Build original-index → output-index for successfully parsed Maven nodes. # deps.dev may emit non-Maven name encodings; skipping those shifts indices, # so edges must be remapped. Truncation drops nodes beyond the fan-out cap. orig_to_out: Dict[int, int] = {} out_nodes: List[Dict[str, Any]] = [] truncated = False for i, raw in enumerate(raw_nodes): if len(out_nodes) >= MAX_TRANSITIVE_GRAPH_NODES: truncated = True break vk = raw.get("versionKey") or {} split = _split_maven_package_name(vk.get("name") or "") if not split: continue gid, aid = split node_errors = raw.get("errors") or [] if not isinstance(node_errors, list): node_errors = [str(node_errors)] orig_to_out[i] = len(out_nodes) out_nodes.append({ "groupId": gid, "artifactId": aid, "version": vk.get("version") or "", "relation": raw.get("relation") or "", "errors": [str(e) for e in node_errors if e], }) if len(raw_nodes) > MAX_TRANSITIVE_GRAPH_NODES: truncated = True edges: List[Dict[str, Any]] = [] for raw in raw_edges: try: frm = int(raw.get("fromNode")) to = int(raw.get("toNode")) except (TypeError, ValueError): continue if frm not in orig_to_out or to not in orig_to_out: if frm >= MAX_TRANSITIVE_GRAPH_NODES or to >= MAX_TRANSITIVE_GRAPH_NODES: truncated = True continue edges.append({ "from": orig_to_out[frm], "to": orig_to_out[to], "requirement": raw.get("requirement") or "", }) out.update({ "ok": True, "error": None, "graphError": graph_error, "nodes": out_nodes, "edges": edges, "partial": bool(truncated or graph_error or any(n.get("errors") for n in out_nodes)), "truncated": truncated, }) return out def _edge_depth_map(edges: List[Dict], root_index: int = 0) -> Dict[int, int]: """BFS depth from ``root_index`` over directed ``from → to`` edges. Used for Maven nearest-wins mediation: the shallowest path wins. Nodes unreachable from the root get no entry (callers treat them as infinite depth). """ adj: Dict[int, List[int]] = {} for e in edges: adj.setdefault(e["from"], []).append(e["to"]) depths: Dict[int, int] = {root_index: 0} queue: List[int] = [root_index] head = 0 while head < len(queue): cur = queue[head] head += 1 for nxt in adj.get(cur, []): if nxt in depths: continue depths[nxt] = depths[cur] + 1 queue.append(nxt) return depths def resolve_conflict_version( versions: List[str], strategy: str, *, depths_by_version: Optional[Dict[str, int]] = None, ) -> Optional[str]: """Pick the version a build system would mediate to among ``versions``. - ``highest-wins`` (Gradle): highest by ``compare_versions``. - ``nearest-wins`` (Maven): shallowest depth wins; ties broken by highest version (Maven's actual tie-break is declaration order, which we do not have — documented limitation). """ uniq = sorted(set(v for v in versions if v), key=functools.cmp_to_key(compare_versions)) if not uniq: return None if strategy == "highest-wins": return uniq[-1] # nearest-wins if not depths_by_version: # No depth info — fall back to highest (honest degrade). return uniq[-1] best_v = None best_depth = None for v in uniq: d = depths_by_version.get(v) if d is None: continue if best_depth is None or d < best_depth or ( d == best_depth and compare_versions(v, best_v or "") > 0 ): best_v = v best_depth = d return best_v if best_v is not None else uniq[-1] def _conflict_risk(versions: List[str], resolved_to: Optional[str]) -> str: """Heuristic risk label for a multi-version GA conflict. - ``high``: major-version divergence among candidates, or resolved version is not the highest (classic nearest-wins silent downgrade). - ``medium``: minor divergence. - ``low``: patch-only / unknown. """ if not versions or not resolved_to: return "low" uniq = list(dict.fromkeys(versions)) highest = max(uniq, key=functools.cmp_to_key(compare_versions)) if resolved_to != highest: return "high" kinds = set() for v in uniq: if v == resolved_to: continue kinds.add(get_upgrade_type(v, resolved_to) if compare_versions(resolved_to, v) > 0 else get_upgrade_type(resolved_to, v)) if "major" in kinds: return "high" if "minor" in kinds: return "medium" return "low" def strategy_for_build_system(build_system: str) -> str: """Map detected build system to mediation strategy label.""" if build_system == "maven": return "nearest-wins" # Gradle (and unknown → Gradle-like highest-wins as the safer default for # modern JVM builds; callers can override via buildSystem arg). return "highest-wins" def get_transitive_graph( group_id: str, artifact_id: str, version: str, ) -> Dict[str, Any]: """Resolved transitive graph for one GAV via deps.dev (#287).""" fetched = fetch_depsdev_dependencies(group_id, artifact_id, version) result: Dict[str, Any] = { "groupId": group_id, "artifactId": artifact_id, "version": version, "nodes": [ {"groupId": n["groupId"], "artifactId": n["artifactId"], "version": n["version"]} for n in fetched["nodes"] ], "edges": [ {"from": e["from"], "to": e["to"]} for e in fetched["edges"] ], "partial": fetched["partial"], "truncated": fetched["truncated"], } if fetched.get("graphError"): result["graphError"] = fetched["graphError"] if fetched.get("capabilityUnavailable"): result["capabilityUnavailable"] = fetched["capabilityUnavailable"] if not fetched["ok"]: result["partial"] = True result["error"] = fetched.get("error") or "deps.dev unavailable" # Surface node-level errors when present even on ok path above; here # the graph is empty. else: # Attach per-node errors only when any exist (keep happy-path lean). node_errors = [] for i, n in enumerate(fetched["nodes"]): if n.get("errors"): node_errors.append({"index": i, "errors": n["errors"]}) if node_errors: result["nodeErrors"] = node_errors[:MAX_DEPSDEV_ERRORS_REPORTED] return result def _bfs_predecessors(edges: List[Dict[str, int]], root_index: int) -> Dict[int, int]: """BFS predecessor map from ``root_index`` over directed ``from -> to`` edges. Same traversal shape as ``_edge_depth_map`` (adjacency built the same way) but returns predecessors for shortest-path reconstruction instead of depths. Nodes unreachable from the root get no entry. """ adj: Dict[int, List[int]] = {} for e in edges: adj.setdefault(e["from"], []).append(e["to"]) predecessor: Dict[int, int] = {} visited = {root_index} queue: List[int] = [root_index] head = 0 while head < len(queue): cur = queue[head] head += 1 for nxt in adj.get(cur, []): if nxt in visited: continue visited.add(nxt) predecessor[nxt] = cur queue.append(nxt) return predecessor def _reconstruct_path( predecessor: Dict[int, int], root_index: int, target_index: int, ) -> Optional[List[int]]: """Node-index path from root to target (inclusive, root-first). Returns ``None`` when ``target_index`` is not the root and has no predecessor (unreachable from root). """ if target_index != root_index and target_index not in predecessor: return None path = [target_index] while path[-1] != root_index: path.append(predecessor[path[-1]]) path.reverse() return path def get_vulnerability_paths(group_id: str, artifact_id: str, version: str) -> Dict[str, Any]: """Dependency path from a project root GAV to each vulnerable transitive node (#413). Fetches the deps.dev transitive graph ONCE via ``fetch_depsdev_dependencies`` — the same cached call ``get_transitive_graph`` wraps, called directly here rather than through that wrapper because the wrapper's trimmed node shape drops the ``relation`` field this needs to locate the root — dedupes nodes by ``groupId:artifactId:version``, batches every unique GAV through ``query_osv_batch`` ONCE, then BFS's from the root over the ``edges`` index-pairs to find the SHORTEST path to every node with >=1 known vulnerability. Root identification: the node whose ``relation`` is ``SELF``. deps.dev sometimes leaves ``relation`` empty on an untagged first node (the same quirk ``check_license_compliance`` documents and falls back on), so an unmatched relation falls back to node 0. Never raises for network/HTTP/parse failures — degrades like the underlying graph/OSV calls (``partial`` / ``capabilityUnavailable``). """ fetched = fetch_depsdev_dependencies(group_id, artifact_id, version) result: Dict[str, Any] = { "groupId": group_id, "artifactId": artifact_id, "version": version, "vulnerabilityPaths": [], "partial": fetched.get("partial", True), "truncated": fetched.get("truncated", False), } if fetched.get("capabilityUnavailable"): result["capabilityUnavailable"] = fetched["capabilityUnavailable"] if fetched.get("graphError"): result["graphError"] = fetched["graphError"] if not fetched.get("ok"): result["partial"] = True result["error"] = fetched.get("error") or "deps.dev unavailable" return result nodes = fetched.get("nodes") or [] edges = fetched.get("edges") or [] if not nodes: return result root_index = 0 for i, n in enumerate(nodes): if (n.get("relation") or "").upper() == "SELF": root_index = i break # Dedupe nodes by GAV: one OSV query per unique coordinate, even if (as a # defensive measure only — deps.dev's own graph model already represents a # shared dependency as one node with multiple incoming edges, never as two # nodes) the same GAV somehow appears at more than one node index. key_to_indices: Dict[str, List[int]] = {} unique_gavs: List[Dict[str, str]] = [] key_order: List[str] = [] for i, n in enumerate(nodes): g = n.get("groupId") or "" a = n.get("artifactId") or "" v = n.get("version") or "" if not g or not a or not v: continue key = f"{g}:{a}:{v}" if key not in key_to_indices: key_to_indices[key] = [] unique_gavs.append({"groupId": g, "artifactId": a, "version": v}) key_order.append(key) key_to_indices[key].append(i) if not unique_gavs: return result notes: List[str] = [] if len(unique_gavs) > MAX_VULN_DEPENDENCIES: unique_gavs = unique_gavs[:MAX_VULN_DEPENDENCIES] key_order = key_order[:MAX_VULN_DEPENDENCIES] result["partial"] = True notes.append( f"Unique dependencies truncated to {MAX_VULN_DEPENDENCIES} " f"(MAX_VULN_DEPENDENCIES) before querying OSV; results are partial." ) key_to_gav = dict(zip(key_order, unique_gavs)) osv_results = query_osv_batch(unique_gavs) vulns_by_key: Dict[str, List[Dict[str, Any]]] = {} osv_capability: Optional[str] = None for key, entry in zip(key_order, osv_results): cap = entry.get("capabilityUnavailable") if cap and osv_capability is None: osv_capability = cap vulns = entry.get("vulnerabilities") or [] if vulns: vulns_by_key[key] = vulns if osv_capability and not result.get("capabilityUnavailable"): result["capabilityUnavailable"] = osv_capability if notes: result["notes"] = notes if not vulns_by_key: return result predecessor = _bfs_predecessors(edges, root_index) hits: List[Dict[str, Any]] = [] unreachable: List[Dict[str, Any]] = [] for key in key_order: vulns = vulns_by_key.get(key) if not vulns: continue gav = key_to_gav[key] best_path: Optional[List[int]] = None for idx in key_to_indices[key]: path = _reconstruct_path(predecessor, root_index, idx) if path is not None and (best_path is None or len(path) < len(best_path)): best_path = path if best_path is None: unreachable.append({"vulnerableNode": gav, "vulnerabilities": vulns}) continue hits.append({ "vulnerableNode": gav, "path": [ { "groupId": nodes[p].get("groupId") or "", "artifactId": nodes[p].get("artifactId") or "", "version": nodes[p].get("version") or "", } for p in best_path ], "vulnerabilities": vulns, }) result["vulnerabilityPaths"] = hits if unreachable: result["unreachableVulnerabilities"] = unreachable result["partial"] = True return result # Display cap for a closure path. Longer paths keep the first three GAVs and # the leaf; the leaf is the classified coordinate and is never dropped. _UPGRADE_PATH_GAVS = 8 _UPGRADE_PATH_NONE = 10 ** 9 def _is_closure_gav(node: Dict[str, Any]) -> bool: """True for a module coordinate. Project components and synthetic roots are not.""" if node.get("synthetic") or node.get("project"): return False return bool(node.get("groupId") and node.get("artifactId") and node.get("version")) def _sorted_versions(versions: List[str]) -> List[str]: seen = set() uniq: List[str] = [] for version in versions: if not version or version in seen: continue seen.add(version) uniq.append(version) return sorted(uniq, key=functools.cmp_to_key(compare_versions)) def _dedup_usages(usages: List[Dict[str, Any]]) -> List[Dict[str, Any]]: out: List[Dict[str, Any]] = [] seen = set() for usage in usages: key = (usage.get("module"), usage.get("configuration"), usage.get("version")) if key in seen: continue seen.add(key) out.append({ "module": usage.get("module"), "configuration": usage.get("configuration"), "version": usage.get("version"), }) return out def _empty_upgrade_dependencies() -> Dict[str, List[Any]]: return {"added": [], "changed": [], "removed": []} def _empty_upgrade_summary() -> Dict[str, int]: return {"added": 0, "changed": 0, "removed": 0, "unchanged": 0} def _empty_vuln_buckets() -> Dict[str, List[Any]]: return {"introduced": [], "remaining": [], "fixed": [], "uncompared": []} def _normalise_closure_graph(trees: List[Dict[str, Any]]) -> Dict[str, Any]: """Merge per-configuration trees into one graph keyed by GAV. Edges address ``node["index"]`` when that field is set, otherwise the list position. A ``project: true`` node (or a node with no GAV) is dropped; its children attach to the nearest kept ancestor, or to a synthetic ``{synthetic, module, configuration}`` root when every ancestor is a project component. The same GAV in two trees is one node; each tree adds a ``{module, configuration, version}`` usage. Duplicate edges collapse. """ nodes_out: List[Dict[str, Any]] = [] edges_out: List[Dict[str, int]] = [] gav_index: Dict[Tuple[str, str, str], int] = {} synthetic_index: Dict[Tuple[str, str], int] = {} edge_seen = set() def ensure_gav(node: Dict[str, Any], module: Any, configuration: Any) -> int: key = (node["groupId"], node["artifactId"], node["version"]) if key not in gav_index: gav_index[key] = len(nodes_out) nodes_out.append({ "groupId": key[0], "artifactId": key[1], "version": key[2], "usages": [], }) usage = {"module": module, "configuration": configuration, "version": key[2]} usages = nodes_out[gav_index[key]]["usages"] if usage not in usages: usages.append(usage) return gav_index[key] def ensure_synthetic(module: Any, configuration: Any) -> int: key = (module, configuration) if key not in synthetic_index: synthetic_index[key] = len(nodes_out) nodes_out.append({ "synthetic": True, "module": module, "configuration": configuration, }) return synthetic_index[key] def add_edge(src: int, dst: int) -> None: if src == dst or (src, dst) in edge_seen: return edge_seen.add((src, dst)) edges_out.append({"from": src, "to": dst}) for tree in trees or []: module = tree.get("module") configuration = tree.get("configuration") local_nodes = list(tree.get("nodes") or []) local_edges = list(tree.get("edges") or []) index_to_pos: Dict[int, int] = {} for pos, node in enumerate(local_nodes): raw_index = node.get("index") idx = raw_index if isinstance(raw_index, int) else pos index_to_pos[idx] = pos parents: Dict[int, List[int]] = {} positioned: List[Tuple[int, int]] = [] for edge in local_edges: src_raw = edge.get("from") dst_raw = edge.get("to") if not isinstance(src_raw, int) or not isinstance(dst_raw, int): continue src_pos = index_to_pos.get(src_raw) dst_pos = index_to_pos.get(dst_raw) if src_pos is None or dst_pos is None: continue positioned.append((src_pos, dst_pos)) parents.setdefault(dst_pos, []).append(src_pos) memo: Dict[int, List[int]] = {} def kept_ancestors(idx: int, trail: Tuple[int, ...]) -> List[int]: # A removed project component keeps every parent edge. One memoized # ancestor would drop the other parent and hide the shorter path. if idx in memo: return memo[idx] if idx in trail: return [] node = local_nodes[idx] if node.get("synthetic"): merged = [ensure_synthetic( node.get("module", module), node.get("configuration", configuration), )] memo[idx] = merged return merged if _is_closure_gav(node): merged = [ensure_gav(node, module, configuration)] memo[idx] = merged return merged ancestors: List[int] = [] seen_anc = set() for parent in parents.get(idx, []): for ancestor in kept_ancestors(parent, trail + (idx,)): if ancestor not in seen_anc: seen_anc.add(ancestor) ancestors.append(ancestor) if not ancestors: ancestors = [ensure_synthetic(module, configuration)] memo[idx] = ancestors return ancestors for pos, node in enumerate(local_nodes): if _is_closure_gav(node): kept_ancestors(pos, ()) for src_pos, dst_pos in positioned: if not _is_closure_gav(local_nodes[dst_pos]): continue dst_ids = kept_ancestors(dst_pos, ()) for src_id in kept_ancestors(src_pos, ()): for dst_id in dst_ids: add_edge(src_id, dst_id) return {"nodes": nodes_out, "edges": edges_out} def _display_closure_path(full: List[Dict[str, Any]]) -> Dict[str, Any]: """Shorten a root-first path to GAV nodes. Synthetics are not displayed. More than ``_UPGRADE_PATH_GAVS`` GAVs keeps the first three and the leaf. ``pathOmitted`` counts dropped middle GAVs plus synthetic hops. ``pathLength`` stays the full node count so ranking does not use the shortened list. """ gavs: List[Dict[str, str]] = [] skipped = 0 for node in full: if node.get("synthetic") or not node.get("groupId"): skipped += 1 continue gavs.append({ "groupId": node["groupId"], "artifactId": node["artifactId"], "version": node["version"], }) omitted = skipped if len(gavs) > _UPGRADE_PATH_GAVS: omitted += len(gavs) - 4 shown = gavs[:3] + [gavs[-1]] else: shown = gavs out: Dict[str, Any] = {"path": shown, "pathLength": len(full)} if omitted: out["pathOmitted"] = omitted return out def _closure_paths( graph: Dict[str, Any], upgrades: List[Dict[str, Any]], *, project_graph: bool = True, ) -> Dict[Tuple[str, str, str], Dict[str, Any]]: """Shortest path from a substituted GA to each GAV node. One ``_bfs_predecessors`` per substituted root (no multi-source mode). A synthetic configuration root that can reach a substituted GA is a fallback root, but only through nodes that have a usage in that same module and configuration. A shared GAV must not carry the walk into another configuration. ``sideEffect`` is omitted when ``project_graph`` is false (deps.dev has no cross-tree effects). """ nodes = list(graph.get("nodes") or []) edges = list(graph.get("edges") or []) sub_indexes: List[int] = [] seen_sub = set() for upgrade in upgrades or []: ga = (upgrade.get("groupId"), upgrade.get("artifactId")) if not ga[0] or not ga[1]: continue for idx, node in enumerate(nodes): if idx in seen_sub: continue if (node.get("groupId"), node.get("artifactId")) == ga: sub_indexes.append(idx) seen_sub.add(idx) best: Dict[int, List[int]] = {} def consider(roots: List[int], walk_edges: List[Dict[str, int]], *, only_missing: bool) -> None: for root in roots: predecessor = _bfs_predecessors(walk_edges, root) for idx, node in enumerate(nodes): if node.get("synthetic") or not _is_closure_gav(node): continue if only_missing and idx in best: continue path = _reconstruct_path(predecessor, root, idx) if path is None: continue previous = best.get(idx) if previous is None or len(path) < len(previous): best[idx] = path def in_config(node: Dict[str, Any], module: Any, configuration: Any) -> bool: if node.get("synthetic"): return node.get("module") == module and node.get("configuration") == configuration for usage in node.get("usages") or []: if usage.get("module") == module and usage.get("configuration") == configuration: return True return False def config_edges(module: Any, configuration: Any, root: int) -> List[Dict[str, int]]: allowed = { idx for idx, node in enumerate(nodes) if in_config(node, module, configuration) } return [ edge for edge in edges if edge["to"] in allowed and (edge["from"] == root or edge["from"] in allowed) ] consider(sub_indexes, edges, only_missing=False) if project_graph and sub_indexes: for idx, node in enumerate(nodes): if not node.get("synthetic"): continue scoped = config_edges(node.get("module"), node.get("configuration"), idx) predecessor = _bfs_predecessors(scoped, idx) # Reachable substituted nodes are in the predecessor map (the root is not). if any(sub in predecessor for sub in sub_indexes): consider([idx], scoped, only_missing=True) out: Dict[Tuple[str, str, str], Dict[str, Any]] = {} for idx, node in enumerate(nodes): if not _is_closure_gav(node): continue key = (node["groupId"], node["artifactId"], node["version"]) info: Dict[str, Any] = {} path_idx = best.get(idx) if path_idx is not None: displayed = _display_closure_path([nodes[i] for i in path_idx]) info["path"] = displayed["path"] info["pathLength"] = displayed["pathLength"] if displayed.get("pathOmitted"): info["pathOmitted"] = displayed["pathOmitted"] if project_graph: info["sideEffect"] = path_idx is None if info: out[key] = info return out def _ga_versions_and_usages( graph: Dict[str, Any], exclude: set, ) -> Tuple[Dict[Tuple[str, str], List[str]], Dict[Tuple[str, str], List[Dict[str, Any]]]]: versions: Dict[Tuple[str, str], List[str]] = {} usages: Dict[Tuple[str, str], List[Dict[str, Any]]] = {} for node in graph.get("nodes") or []: if not _is_closure_gav(node): continue ga = (node["groupId"], node["artifactId"]) if ga in exclude: continue versions.setdefault(ga, []).append(node["version"]) usages.setdefault(ga, []).extend(node.get("usages") or []) return versions, usages def _attach_path( row: Dict[str, Any], side_effect_versions: List[Optional[Dict[str, Any]]], *, project_graph: bool, ) -> int: """Copy the shortest full path onto a diff row. Returns that length for sorting.""" best: Optional[Dict[str, Any]] = None for candidate in side_effect_versions: if not candidate or "pathLength" not in candidate: continue if best is None or candidate["pathLength"] < best["pathLength"]: best = candidate if best and best.get("path"): row["path"] = best["path"] if best.get("pathOmitted"): row["pathOmitted"] = best["pathOmitted"] length = best["pathLength"] if best and "pathLength" in best else _UPGRADE_PATH_NONE if project_graph: reached = any( candidate is not None and candidate.get("sideEffect") is False for candidate in side_effect_versions ) row["sideEffect"] = not reached return length def _diff_closure( before_side: Dict[str, Any], after_side: Dict[str, Any], upgrades: List[Dict[str, Any]], *, project_graph: bool = True, ) -> Dict[str, Any]: """Classify the closure delta by GA. One row per coordinate family. Upgraded direct GAs are removed from both sets before classification and are not added, changed, or removed rows; they stay in the graphs so path search can start there. Either side ``ok: false`` yields empty buckets — that is not an empty project. ``ok: true`` with ``diffReliable: false`` still fills the buckets. """ empty: Dict[str, Any] = { "added": [], "changed": [], "removed": [], "summary": _empty_upgrade_summary(), "dependenciesTruncated": False, "diffReliable": False, "compared": False, } if not before_side.get("ok") or not after_side.get("ok"): return empty exclude = set() for upgrade in upgrades or []: group_id = upgrade.get("groupId") artifact_id = upgrade.get("artifactId") if group_id and artifact_id: exclude.add((group_id, artifact_id)) before_graph = before_side.get("graph") or {"nodes": [], "edges": []} after_graph = after_side.get("graph") or {"nodes": [], "edges": []} before_versions, before_usages = _ga_versions_and_usages(before_graph, exclude) after_versions, after_usages = _ga_versions_and_usages(after_graph, exclude) before_paths = _closure_paths(before_graph, upgrades, project_graph=project_graph) after_paths = _closure_paths(after_graph, upgrades, project_graph=project_graph) added: List[Dict[str, Any]] = [] changed: List[Dict[str, Any]] = [] removed: List[Dict[str, Any]] = [] unchanged = 0 for ga in set(before_versions) | set(after_versions): before_set = _sorted_versions(before_versions.get(ga, [])) after_set = _sorted_versions(after_versions.get(ga, [])) group_id, artifact_id = ga if not before_set and after_set: infos = [after_paths.get((group_id, artifact_id, v)) for v in after_set] row: Dict[str, Any] = { "groupId": group_id, "artifactId": artifact_id, "versions": after_set, } usages = _dedup_usages(after_usages.get(ga, [])) if usages: row["usages"] = usages length = _attach_path(row, infos, project_graph=project_graph) row["_pathLength"] = length added.append(row) elif before_set and not after_set: infos = [before_paths.get((group_id, artifact_id, v)) for v in before_set] row = { "groupId": group_id, "artifactId": artifact_id, "versions": before_set, } usages = _dedup_usages(before_usages.get(ga, [])) if usages: row["usages"] = usages length = _attach_path(row, infos, project_graph=False) row["_pathLength"] = length removed.append(row) elif set(before_set) == set(after_set): unchanged += 1 else: infos = [after_paths.get((group_id, artifact_id, v)) for v in after_set] row = { "groupId": group_id, "artifactId": artifact_id, "fromVersions": before_set, "toVersions": after_set, } usages = _dedup_usages(after_usages.get(ga, []) + before_usages.get(ga, [])) if usages: row["usages"] = usages length = _attach_path(row, infos, project_graph=project_graph) row["_pathLength"] = length changed.append(row) def sort_key(row: Dict[str, Any]) -> Tuple[int, int, str, str]: side = 1 if row.get("sideEffect") else 0 raw_length = row.get("_pathLength", _UPGRADE_PATH_NONE) length = raw_length if isinstance(raw_length, int) else _UPGRADE_PATH_NONE return (side, length, str(row.get("groupId") or ""), str(row.get("artifactId") or "")) added.sort(key=sort_key) changed.sort(key=sort_key) removed.sort(key=sort_key) summary = { "added": len(added), "changed": len(changed), "removed": len(removed), "unchanged": unchanged, } truncated = False def cap(rows: List[Dict[str, Any]]) -> List[Dict[str, Any]]: nonlocal truncated if len(rows) > MAX_UPGRADE_DIFF_ROWS: truncated = True rows = rows[:MAX_UPGRADE_DIFF_ROWS] for row in rows: row.pop("_pathLength", None) return rows reliable = bool(before_side.get("diffReliable", True)) and bool(after_side.get("diffReliable", True)) return { "added": cap(added), "changed": cap(changed), "removed": cap(removed), "summary": summary, "dependenciesTruncated": truncated, "diffReliable": reliable, "compared": True, } def _select_delta_gavs( upgrades: List[Dict[str, Any]], diff: Dict[str, Any], *, cap: int = MAX_VULN_DEPENDENCIES, include_targets: bool = True, ) -> Dict[str, Any]: """Order GAVs for one OSV or license pass. Targets (every ``toVersion``, then every ``fromVersion``) are reserved and never dropped. The closure delta fills the remaining slots: added, changed-after, changed-before, removed. ``truncated`` is true when a closure GAV did not fit. ``include_targets`` false is the license order (the caller applies ``MAX_UPGRADE_LICENSE_NODES``). """ ordered: List[Dict[str, str]] = [] def push(group_id: Any, artifact_id: Any, version: Any) -> None: if not group_id or not artifact_id or not version: return ordered.append({ "groupId": group_id, "artifactId": artifact_id, "version": version, }) if include_targets: for upgrade in upgrades or []: push(upgrade.get("groupId"), upgrade.get("artifactId"), upgrade.get("toVersion")) for upgrade in upgrades or []: push(upgrade.get("groupId"), upgrade.get("artifactId"), upgrade.get("fromVersion")) for upgrade in upgrades or []: # The version Gradle selected can differ from toVersion (not landed). for version in upgrade.get("selectedVersions") or []: push(upgrade.get("groupId"), upgrade.get("artifactId"), version) for group_id, artifact_id, version, _kind in _ordered_license_gavs(diff): push(group_id, artifact_id, version) deduped: List[Dict[str, str]] = [] seen = set() for gav in ordered: key = (gav["groupId"], gav["artifactId"], gav["version"]) if key in seen: continue seen.add(key) deduped.append(gav) if not include_targets: return {"gavs": deduped[:cap], "truncated": len(deduped) > cap} target_keys = set() for upgrade in upgrades or []: for version_key in ("toVersion", "fromVersion"): pinned = upgrade.get(version_key) if upgrade.get("groupId") and upgrade.get("artifactId") and pinned: target_keys.add((upgrade.get("groupId"), upgrade.get("artifactId"), pinned)) for version in upgrade.get("selectedVersions") or []: if upgrade.get("groupId") and upgrade.get("artifactId") and version: target_keys.add((upgrade.get("groupId"), upgrade.get("artifactId"), version)) prefix: List[Dict[str, str]] = [] rest: List[Dict[str, str]] = [] for gav in deduped: key = (gav["groupId"], gav["artifactId"], gav["version"]) if key in target_keys: prefix.append(gav) target_keys.discard(key) else: rest.append(gav) room = max(0, cap - len(prefix)) return {"gavs": prefix + rest[:room], "truncated": len(rest) > room} def _diff_path_fields(diff: Dict[str, Any], group_id: str, artifact_id: str) -> Dict[str, Any]: for bucket in ("added", "changed", "removed"): for row in diff.get(bucket) or []: if row.get("groupId") == group_id and row.get("artifactId") == artifact_id: fields: Dict[str, Any] = {} if row.get("path"): fields["path"] = row["path"] if row.get("pathOmitted"): fields["pathOmitted"] = row["pathOmitted"] return fields return {} def _upgrade_vuln_item( raw: Dict[str, Any], *, relation: str, group_id: str, artifact_id: str, version: str, selected_version: Optional[str], path_fields: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: vuln_id = raw.get("id") or "" if "malicious" in raw: malicious = bool(raw.get("malicious")) else: malicious = _is_malicious_id(vuln_id) item: Dict[str, Any] = { "id": vuln_id, "summary": raw.get("summary") or "", "url": raw.get("url") or "", "malicious": malicious, "relation": relation, "vulnerableNode": { "groupId": group_id, "artifactId": artifact_id, "version": version, }, } if raw.get("severity"): item["severity"] = raw["severity"] fixed = raw.get("fixedVersion") if isinstance(fixed, str) and fixed.strip(): item["fixedVersion"] = fixed if relation in ("introduced", "remaining", "target") and selected_version: item["clearedBySelection"] = compare_versions(selected_version, fixed) >= 0 if relation == "uncompared": item["reason"] = "before-version not queried" if relation != "target" and path_fields: if path_fields.get("path"): item["path"] = path_fields["path"] if path_fields.get("pathOmitted"): item["pathOmitted"] = path_fields["pathOmitted"] return item def _lowest_version_hit( hits: List[Tuple[str, Dict[str, Any]]], ) -> Tuple[str, Dict[str, Any]]: best = hits[0] for candidate in hits[1:]: if compare_versions(candidate[0], best[0]) < 0: best = candidate return best def _vulns_by_version( records_by_version: Dict[str, Dict[str, Any]], versions: List[str], ) -> Dict[str, List[Tuple[str, Dict[str, Any]]]]: found: Dict[str, List[Tuple[str, Dict[str, Any]]]] = {} for version in versions: record = records_by_version.get(version) if not record: continue for vuln in record.get("vulnerabilities") or []: vuln_id = vuln.get("id") or "" if not vuln_id: continue found.setdefault(vuln_id, []).append((version, vuln)) return found def _classify_vuln_delta( upgrades: List[Dict[str, Any]], diff: Dict[str, Any], records: List[Dict[str, Any]], ) -> Dict[str, Any]: """Classify already-fetched OSV records. Does not query the network. A before-version that was not queried does not produce ``introduced`` or ``fixed``; the after-version ids go to ``uncompared`` and ``fixesIncomplete`` is set. A ``MAL-`` on a queried after-version stays ``malicious`` and is not labeled introduced. ``safeUpgrade`` is computed only for a target ``toVersion`` list, and omitted when that list is empty. """ queried: Dict[Tuple[str, str, str], Dict[str, Any]] = {} capability: Optional[str] = None for record in records or []: group_id = record.get("groupId") artifact_id = record.get("artifactId") version = record.get("version") if not group_id or not artifact_id or not version: continue key = (group_id, artifact_id, version) if key not in queried: queried[key] = record if record.get("capabilityUnavailable") and capability is None: capability = record["capabilityUnavailable"] by_ga: Dict[Tuple[str, str], Dict[str, Dict[str, Any]]] = {} for (group_id, artifact_id, version), record in queried.items(): by_ga.setdefault((group_id, artifact_id), {})[version] = record introduced: List[Dict[str, Any]] = [] remaining: List[Dict[str, Any]] = [] fixed: List[Dict[str, Any]] = [] uncompared: List[Dict[str, Any]] = [] fixes_incomplete = False input_truncated = False def missing(group_id: str, artifact_id: str, versions: List[str]) -> List[str]: return [ version for version in versions if (group_id, artifact_id, version) not in queried ] def classify_pair( group_id: str, artifact_id: str, before_versions: List[str], after_versions: List[str], *, target: bool, ) -> List[Dict[str, Any]]: """Return introduced/remaining/fixed items. Closure buckets are filled only when ``target`` is false.""" nonlocal fixes_incomplete, input_truncated delta: List[Dict[str, Any]] = [] before_missing = missing(group_id, artifact_id, before_versions) after_missing = missing(group_id, artifact_id, after_versions) if after_missing and after_versions: input_truncated = True path_fields = None if target else _diff_path_fields(diff, group_id, artifact_id) versions = by_ga.get((group_id, artifact_id), {}) queried_after = [ version for version in after_versions if (group_id, artifact_id, version) in queried ] if before_missing and before_versions: fixes_incomplete = True if not target: # A missing before-version blocks introduced/fixed, but a MAL- # on a queried after-version still has to be visible to the ranker. after_hits = _vulns_by_version(versions, queried_after) for vuln_id in sorted(after_hits): version, raw = _lowest_version_hit(after_hits[vuln_id]) uncompared.append(_upgrade_vuln_item( raw, relation="uncompared", group_id=group_id, artifact_id=artifact_id, version=version, selected_version=version, path_fields=path_fields, )) return delta before_hits = _vulns_by_version(versions, before_versions) after_hits = _vulns_by_version(versions, queried_after) for vuln_id in sorted(set(before_hits) | set(after_hits)): on_before = vuln_id in before_hits on_after = vuln_id in after_hits if on_before and on_after: relation = "remaining" version, raw = _lowest_version_hit(after_hits[vuln_id]) elif on_after: relation = "introduced" version, raw = _lowest_version_hit(after_hits[vuln_id]) elif after_missing: # Another after-version was not queried, so this id may still be selected. continue else: relation = "fixed" version, raw = _lowest_version_hit(before_hits[vuln_id]) if after_versions: # GA still selected: point at the after coordinate. version = _sorted_versions(after_versions)[0] item = _upgrade_vuln_item( raw, relation=relation, group_id=group_id, artifact_id=artifact_id, version=version, selected_version=version if relation != "fixed" else None, path_fields=path_fields, ) delta.append(item) if not target: {"introduced": introduced, "remaining": remaining, "fixed": fixed}[relation].append(item) return delta targets: List[Dict[str, Any]] = [] for upgrade in upgrades or []: group_id = upgrade.get("groupId") or "" artifact_id = upgrade.get("artifactId") or "" from_version = upgrade.get("fromVersion") or "" to_version = upgrade.get("toVersion") or "" target: Dict[str, Any] = { "groupId": group_id, "artifactId": artifact_id, "fromVersion": from_version, "toVersion": to_version, } for field in ( "selectedVersions", "landed", "substitutionMatched", "rewroteVersionless", "targetPresent", ): if field in upgrade: target[field] = upgrade[field] selected_versions = [ version for version in (upgrade.get("selectedVersions") or []) if version ] # When Gradle selected something other than toVersion, that version # is what the build will use. vuln_versions = selected_versions or ([to_version] if to_version else []) raw_vulns = [] saw_record = False for version in vuln_versions: record: Optional[Dict[str, Any]] = queried.get((group_id, artifact_id, version)) if record is None: input_truncated = True continue saw_record = True if record.get("capabilityUnavailable") and "capabilityUnavailable" not in target: target["capabilityUnavailable"] = record["capabilityUnavailable"] for raw in record.get("vulnerabilities") or []: if raw.get("id"): raw_vulns.append((version, raw)) if saw_record: target["vulnerabilities"] = [ _upgrade_vuln_item( raw, relation="target", group_id=group_id, artifact_id=artifact_id, version=version, selected_version=version, ) for version, raw in raw_vulns ] if raw_vulns: safe = _compute_safe_upgrade([raw for _version, raw in raw_vulns]) if safe is not None: target["safeUpgrade"] = safe elif to_version: input_truncated = True to_record = queried.get((group_id, artifact_id, to_version)) if to_version else None if from_version and to_version: after_versions = selected_versions or [to_version] delta = classify_pair( group_id, artifact_id, [from_version], after_versions, target=True, ) if (group_id, artifact_id, from_version) in queried and to_record is not None: target["vulnerabilityDelta"] = delta targets.append(target) for row in diff.get("added") or []: classify_pair( row["groupId"], row["artifactId"], [], list(row.get("versions") or []), target=False, ) for row in diff.get("changed") or []: classify_pair( row["groupId"], row["artifactId"], list(row.get("fromVersions") or []), list(row.get("toVersions") or []), target=False, ) for row in diff.get("removed") or []: classify_pair( row["groupId"], row["artifactId"], list(row.get("versions") or []), [], target=False, ) def sort_items(items: List[Dict[str, Any]]) -> List[Dict[str, Any]]: return sorted(items, key=lambda item: ( item["vulnerableNode"]["groupId"], item["vulnerableNode"]["artifactId"], item["vulnerableNode"]["version"], item["id"], )) result: Dict[str, Any] = { "targets": targets, "vulnerabilities": { "introduced": sort_items(introduced), "remaining": sort_items(remaining), "fixed": sort_items(fixed), "uncompared": sort_items(uncompared), }, "fixesIncomplete": fixes_incomplete, "inputTruncated": input_truncated, } if capability: result["capabilityUnavailable"] = capability return result def _ordered_license_gavs(diff: Dict[str, Any]) -> List[Tuple[str, str, str, str]]: """Added, changed-after, changed-before, removed. Targets are not in ``diff``.""" ordered: List[Tuple[str, str, str, str]] = [] for row in diff.get("added") or []: for version in row.get("versions") or []: ordered.append((row["groupId"], row["artifactId"], version, "added")) for row in diff.get("changed") or []: for version in row.get("toVersions") or []: ordered.append((row["groupId"], row["artifactId"], version, "changed-to")) for row in diff.get("changed") or []: for version in row.get("fromVersions") or []: ordered.append((row["groupId"], row["artifactId"], version, "changed-from")) for row in diff.get("removed") or []: for version in row.get("versions") or []: ordered.append((row["groupId"], row["artifactId"], version, "removed")) return ordered def _license_delta( diff: Dict[str, Any], fetched: List[Dict[str, Any]], *, project_license: Optional[str] = None, disallow: Optional[List[str]] = None, ) -> Dict[str, Any]: """License posture for an already-fetched closure delta. No network. ``categoriesIntroduced`` is omitted unless every selected GAV, including changed-before and removed, was fetched with ``ok: true``. A verdict is attached to appeared rows only. """ policy = resolve_license_policy(project_license, disallow) by_gav: Dict[Tuple[str, str, str], Dict[str, Any]] = {} capability: Optional[str] = None for record in fetched or []: group_id = record.get("groupId") artifact_id = record.get("artifactId") version = record.get("version") if not group_id or not artifact_id or not version: continue key = (group_id, artifact_id, version) if key not in by_gav: by_gav[key] = record if record.get("capabilityUnavailable") and capability is None: capability = record["capabilityUnavailable"] appeared: List[Dict[str, Any]] = [] disappeared: List[Dict[str, Any]] = [] selected = _ordered_license_gavs(diff) complete = True for group_id, artifact_id, version, kind in selected: hit = by_gav.get((group_id, artifact_id, version)) if not hit or not hit.get("ok"): if hit and hit.get("status") == 404: # A missing record is a review verdict, not a silent skip # and not a transport failure. base = { "groupId": group_id, "artifactId": artifact_id, "version": version, "spdxId": None, "category": "unknown", } if kind in ("added", "changed-to"): verdict = license_compliance_verdict( spdx_id=None, category="unknown", policy=policy, missing_license=True, ) row = dict(base) row["verdict"] = verdict["verdict"] row["reason"] = verdict["reason"] appeared.append(row) else: disappeared.append(base) continue complete = False continue primary = _primary_license_from_depsdev(list(hit.get("licenses") or [])) raw_spdx = primary.get("spdxId") spdx_id = raw_spdx if isinstance(raw_spdx, str) else None raw_category = primary.get("category") category = raw_category if isinstance(raw_category, str) and raw_category else "unknown" base: Dict[str, Any] = { "groupId": group_id, "artifactId": artifact_id, "version": version, "spdxId": spdx_id, "category": category, } if kind in ("added", "changed-to"): verdict = license_compliance_verdict( spdx_id=spdx_id, category=category, policy=policy, missing_license=not spdx_id, ) row = dict(base) row["verdict"] = verdict["verdict"] row["reason"] = verdict["reason"] appeared.append(row) else: disappeared.append(base) result: Dict[str, Any] = {"appeared": appeared, "disappeared": disappeared} if complete and selected: gone = {row["category"] for row in disappeared} introduced: List[str] = [] for row in appeared: row_category = row.get("category") if not isinstance(row_category, str): continue if row_category not in gone and row_category not in introduced: introduced.append(row_category) result["categoriesIntroduced"] = introduced if capability: result["capabilityUnavailable"] = capability return result _ADVISORY_NONE_NOTE = ( "Unchanged transitives were not queried, OSV coverage is incomplete, " "and an empty closure delta is not a safety guarantee." ) _ADVISORY_IDENTITY_NOTE = ( "Nothing was requested to change and the current coordinate was not re-queried." ) _HIGH_SEVERITIES = frozenset({"CRITICAL", "HIGH"}) _INFO_SEVERITIES = frozenset({"MEDIUM", "LOW"}) def _severity_label(item: Dict[str, Any]) -> Optional[str]: severity = item.get("severity") if not severity: return None return str(severity).upper() def _advisory_for_upgrade( *, all_identity: bool = False, targets: Optional[List[Dict[str, Any]]] = None, before_ok: bool = True, after_ok: bool = True, diff_reliable: bool = True, diff: Optional[Dict[str, Any]] = None, vulnerabilities: Optional[Dict[str, Any]] = None, license_delta: Optional[Dict[str, Any]] = None, fixes_incomplete: bool = False, capability_unavailable: Optional[str] = None, input_truncated: bool = False, ) -> Dict[str, Any]: """Rank one upgrade preview. Not a permission and not a safety verdict. ``all_identity`` is the only ``none`` that does not read ``diff_reliable`` or any OSV payload. Targets on that path omit ``vulnerabilities`` (absent, not an empty list). Any other call with ``diff_reliable`` false or no OSV payload does not return ``none``. """ if all_identity: # Return before consulting diff reliability or vulnerability records. echoed: List[Dict[str, Any]] = [] for target in targets or []: echoed.append({ "groupId": target.get("groupId"), "artifactId": target.get("artifactId"), "fromVersion": target.get("fromVersion"), "toVersion": target.get("toVersion"), }) return { "advisory": "none", "diffReliable": False, "partial": False, "notes": [_ADVISORY_IDENTITY_NOTE], "targets": echoed, "dependencies": _empty_upgrade_dependencies(), "summary": _empty_upgrade_summary(), "vulnerabilities": _empty_vuln_buckets(), } if not before_ok or not after_ok: return { "advisory": "unknown", "diffReliable": False, "partial": True, "notes": ["The closure was not compared because a graph side failed."], "targets": list(targets or []), "dependencies": _empty_upgrade_dependencies(), "summary": _empty_upgrade_summary(), "vulnerabilities": _empty_vuln_buckets(), } dependencies = { "added": list((diff or {}).get("added") or []), "changed": list((diff or {}).get("changed") or []), "removed": list((diff or {}).get("removed") or []), } summary = dict((diff or {}).get("summary") or _empty_upgrade_summary()) vuln_buckets = { "introduced": list((vulnerabilities or {}).get("introduced") or []), "remaining": list((vulnerabilities or {}).get("remaining") or []), "fixed": list((vulnerabilities or {}).get("fixed") or []), "uncompared": list((vulnerabilities or {}).get("uncompared") or []), } has_osv_payload = vulnerabilities is not None dependencies_truncated = bool(diff and diff.get("dependenciesTruncated")) license_capability = (license_delta or {}).get("capabilityUnavailable") license_truncated = bool((license_delta or {}).get("truncated")) target_capability = None for target in targets or []: if target.get("capabilityUnavailable"): target_capability = target["capabilityUnavailable"] break capability = capability_unavailable or license_capability or target_capability appeared = list((license_delta or {}).get("appeared") or []) disappeared = list((license_delta or {}).get("disappeared") or []) review_notes: List[str] = [] info_notes: List[str] = [] def target_vulns() -> List[Dict[str, Any]]: rows: List[Dict[str, Any]] = [] for target in targets or []: rows.extend(target.get("vulnerabilities") or []) return rows stop = False for item in target_vulns() + vuln_buckets["introduced"] + vuln_buckets["remaining"] + vuln_buckets["uncompared"]: if item.get("malicious"): stop = True break if not diff_reliable: review_notes.append("diffReliable is false; the diff is not a complete closure.") if fixes_incomplete or vuln_buckets["uncompared"]: review_notes.append("A before-version was not queried; fixes are incomplete.") if capability: review_notes.append("A required OSV or license fetch was unavailable.") if input_truncated or dependencies_truncated: review_notes.append("The closure diff or the OSV input was truncated.") if license_truncated: review_notes.append( "License lookups were capped; some changed coordinates were not licensed." ) for target in targets or []: if target.get("landed") is False: review_notes.append("A target did not land on toVersion.") break for target in targets or []: if target.get("rewroteVersionless"): review_notes.append("A versionless request was rewritten.") break for item in target_vulns(): severity = _severity_label(item) if severity in _HIGH_SEVERITIES or severity is None: review_notes.append("A target toVersion has a critical, high, or unranked vulnerability.") break for item in vuln_buckets["introduced"]: severity = _severity_label(item) if severity in _HIGH_SEVERITIES or severity is None: review_notes.append("An introduced vulnerability is critical, high, or unranked.") break for item in vuln_buckets["remaining"]: severity = _severity_label(item) if severity in _HIGH_SEVERITIES or severity is None: review_notes.append("A remaining vulnerability is critical, high, or unranked.") break if any(row.get("verdict") == "violation" for row in appeared): review_notes.append("An appeared coordinate violates the license policy.") delta_rows = bool(dependencies["added"] or dependencies["changed"] or dependencies["removed"]) if delta_rows: info_notes.append("The closure added, changed, or removed coordinates.") target_fixed = any( item.get("relation") == "fixed" for target in targets or [] for item in target.get("vulnerabilityDelta") or [] ) if vuln_buckets["fixed"] or target_fixed: info_notes.append("A vulnerability is no longer on the selected coordinate.") if any(_severity_label(item) in _INFO_SEVERITIES for item in vuln_buckets["remaining"]): info_notes.append("A remaining vulnerability is medium or low.") if any(_severity_label(item) in _INFO_SEVERITIES for item in vuln_buckets["introduced"]): info_notes.append("An introduced vulnerability is medium or low.") if any(_severity_label(item) in _INFO_SEVERITIES for item in target_vulns()): info_notes.append("A target toVersion vulnerability is medium or low.") if any(row.get("verdict") in ("ok", "review") for row in appeared): info_notes.append("A license row is ok or needs review.") targets_clean = has_osv_payload if targets_clean: for target in targets or []: if "vulnerabilities" not in target or target.get("vulnerabilities") or target.get("capabilityUnavailable"): targets_clean = False break licenses_quiet = license_delta is None or (not appeared and not disappeared) can_be_none = ( diff_reliable and not review_notes and not info_notes and targets_clean and licenses_quiet and not capability and not fixes_incomplete and not input_truncated and not dependencies_truncated and not license_truncated and not delta_rows ) if stop: advisory = "stop" notes = ["A queried after-coordinate is flagged malicious."] elif review_notes: advisory = "review" notes = review_notes elif info_notes: advisory = "info" notes = info_notes elif can_be_none: advisory = "none" notes = [ "No coordinate change outside the target.", _ADVISORY_NONE_NOTE, ] else: # Missing OSV payload, or some other gap, must not look like a clean compare. advisory = "review" notes = ["OSV was not queried for the target, so this is not a clean compare."] partial = advisory == "unknown" or bool( not diff_reliable or fixes_incomplete or capability or input_truncated or dependencies_truncated or license_truncated ) result: Dict[str, Any] = { "advisory": advisory, "diffReliable": diff_reliable, "partial": partial, "notes": notes, "targets": list(targets or []), "dependencies": dependencies, "summary": summary, "vulnerabilities": vuln_buckets if has_osv_payload else _empty_vuln_buckets(), } if fixes_incomplete: result["fixesIncomplete"] = True if dependencies_truncated: result["dependenciesTruncated"] = True if input_truncated: result["inputTruncated"] = True if capability: result["capabilityUnavailable"] = capability if license_delta is not None: result["license"] = license_delta return result # auto resolves to gradle when gradlew exists, else one deps.dev upgrade. # Coordinates are checked after selection and before any resolve. _UPGRADE_GA_RE = re.compile(r"^[A-Za-z0-9_.\-]+$") _UPGRADE_VERSION_RE = re.compile(r"^[A-Za-z0-9_.+\-]+$") _UPGRADE_CHANGED_ONLY_LICENSE_NOTE = ( "Only the changed coordinates were licensed, not the unchanged closure." ) _GRADLE_LICENSE_METADATA_NOTE = ( "License metadata is not taken from the Gradle resolve; " "a missing deps.dev record is review, not a known license." ) _VERSIONLESS_ARM_NOTE = ( "The empty-version arm rewrites every versionless request for that " "coordinate on the dumped production-runtime configurations, which is " "wider than one declaration." ) _GRADLE_ERROR_LIMIT = 500 _GRADLE_UPGRADE_TREE = "===MAVEN_MCP_TREE===" _GRADLE_UPGRADE_TREE_END = "===MAVEN_MCP_TREE_END===" _GRADLE_UPGRADE_TRUNCATED = "===MAVEN_MCP_TRUNCATED===" _GRADLE_UPGRADE_CONFIG_ERROR = "===MAVEN_MCP_CONFIG_ERROR===" _GRADLE_UPGRADE_MATCHED = "===MAVEN_MCP_MATCHED===" _GRADLE_UPGRADE_REQUEST = "===MAVEN_MCP_REQUEST===" def _select_upgrade_graph_source(args: Dict, project_path: str) -> str: """Pick ``gradle`` or ``depsdev`` before any resolve. ``auto`` (the default) is Gradle only when the project is a Gradle build and a wrapper exists. A Gradle file without ``gradlew`` still falls through to deps.dev. Nothing here retries a failed Gradle run as deps.dev. """ requested = args.get("graphSource") or "auto" if requested not in ("auto", "depsdev", "gradle"): raise ValueError("graphSource must be auto, depsdev, or gradle") if requested == "depsdev": return "depsdev" if requested == "gradle": return "gradle" # auto: one filesystem check, not one per coordinate. if ( _detect_build_system(project_path) == "gradle" and _find_gradle_wrapper(project_path) ): return "gradle" return "depsdev" def _upgrade_project_path(args: Dict) -> str: raw = args.get("projectPath") or os.getcwd() if not isinstance(raw, str): raise ValueError("projectPath must be a string") return raw def _upgrade_substitution(args: Dict) -> str: """``exact`` or ``module``. Ignored by the deps.dev fetcher, still validated.""" raw = args.get("substitution", "exact") if raw is None: return "exact" if raw not in ("exact", "module"): raise ValueError("substitution must be exact or module") return raw def _parse_upgrade_requests(raw: Any) -> List[Dict[str, str]]: """Validate upgrade coordinates before any network or process launch.""" if not isinstance(raw, list): raise ValueError("upgrades must be a list of coordinates") upgrades: List[Dict[str, str]] = [] for item in raw: if not isinstance(item, dict): raise ValueError("each upgrade must be an object") group_id = item.get("groupId") artifact_id = item.get("artifactId") from_version = item.get("fromVersion") to_version = item.get("toVersion") if ( not isinstance(group_id, str) or not isinstance(artifact_id, str) or not isinstance(from_version, str) or not isinstance(to_version, str) ): raise ValueError( "groupId, artifactId, fromVersion, and toVersion are required" ) if not _UPGRADE_GA_RE.fullmatch(group_id) or not _UPGRADE_GA_RE.fullmatch(artifact_id): raise ValueError("groupId and artifactId must match ^[A-Za-z0-9_.\\-]+$") if ( not _UPGRADE_VERSION_RE.fullmatch(from_version) or not _UPGRADE_VERSION_RE.fullmatch(to_version) ): raise ValueError("fromVersion and toVersion must match ^[A-Za-z0-9_.+\\-]+$") upgrades.append({ "groupId": group_id, "artifactId": artifact_id, "fromVersion": from_version, "toVersion": to_version, }) if len(upgrades) > MAX_UPGRADE_SUBSTITUTIONS: raise ValueError( f"at most {MAX_UPGRADE_SUBSTITUTIONS} upgrades; the list was not truncated" ) seen = set() for upgrade in upgrades: key = (upgrade["groupId"], upgrade["artifactId"]) if key in seen: raise ValueError("duplicate groupId and artifactId") seen.add(key) identities = [upgrade["fromVersion"] == upgrade["toVersion"] for upgrade in upgrades] if identities and any(identities) and not all(identities): raise ValueError("cannot mix an unchanged version with a real bump") if not upgrades: raise ValueError("at least one upgrade is required") return upgrades def _license_note(prefix: str) -> str: for note in _LICENSE_COMPLIANCE_NOTES: if note.startswith(prefix): return note return prefix def _depsdev_closure_notes(include_licenses: bool) -> List[str]: """Isolation paragraph, plus license caveats when that fetch is in scope.""" notes = [_license_note("Graphs are resolved per root in isolation")] if include_licenses: notes.extend(_shared_license_notes()) return notes def _shared_license_notes() -> List[str]: """Heuristic, SPDX, and changed-only sentences used by both graph sources.""" return [ _license_note("License data comes from deps.dev"), _license_note("Verdicts are heuristic"), _license_note("deps.dev may return SPDX"), _UPGRADE_CHANGED_ONLY_LICENSE_NOTE, ] def _gradle_closure_notes(include_licenses: bool) -> List[str]: """Gradle classpath notes. The deps.dev isolation paragraph does not apply.""" if not include_licenses: return [] return _shared_license_notes() + [_GRADLE_LICENSE_METADATA_NOTE] def _log_upgrade_closure( result: Dict[str, Any], *, before: Optional[Dict[str, Any]] = None, after: Optional[Dict[str, Any]] = None, launched_gradle: bool = False, ) -> None: """One line. No transitive versions and no Gradle stderr.""" upgrades = result.get("upgrades") or [] first = upgrades[0] if upgrades else {} targets = result.get("targets") or [] landed = sum(1 for target in targets if target.get("landed") is True) versionless = sum(1 for target in targets if target.get("rewroteVersionless")) before_ok = None if before is None else bool(before.get("ok")) after_ok = None if after is None else bool(after.get("ok")) before_trees = 0 if before is None else len(before.get("trees") or []) after_trees = 0 if after is None else len(after.get("trees") or []) zero_tree = (before is not None and before_trees == 0) or (after is not None and after_trees == 0) before_exit = None if before is None else before.get("exitCode") after_exit = None if after is None else after.get("exitCode") timed_out = before_exit == 124 or after_exit == 124 _logger.info( "compare_upgrade_closure graphSource=%s advisory=%s partial=%s " "diffReliable=%s ok=%s/%s trees=%d/%d zeroTree=%s upgrades=%d first=%s:%s " "landed=%d rewroteVersionless=%d exit=%s/%s timeout=%s noConfigurationCache=%s " "capability=%s", result.get("graphSource"), result.get("advisory"), result.get("partial"), result.get("diffReliable"), before_ok, after_ok, before_trees, after_trees, zero_tree, len(upgrades), first.get("groupId") or "-", first.get("artifactId") or "-", landed, versionless, before_exit, after_exit, timed_out, launched_gradle, result.get("capabilityUnavailable") or "-", ) def _with_upgrade_source( graph_source: str, upgrades_out: List[Dict[str, Any]], ranked: Dict[str, Any], *, error: Optional[str] = None, capability: Optional[str] = None, truncated: bool = False, notes: Optional[List[str]] = None, ) -> Dict[str, Any]: result = dict(ranked) result["graphSource"] = graph_source result["upgrades"] = upgrades_out if notes: merged = list(result.get("notes") or []) for note in notes: if note not in merged: merged.append(note) result["notes"] = merged if not result.get("notes"): result["notes"] = ["The upgrade closure check produced no notes."] if error: result["error"] = error if capability and not result.get("capabilityUnavailable"): result["capabilityUnavailable"] = capability if truncated: result["truncated"] = True return result def _marker_upgrade_result( graph_source: str, upgrades_out: List[Dict[str, Any]], ) -> Dict[str, Any]: echoed: List[Dict[str, Any]] = [] for upgrade in upgrades_out: row = dict(upgrade) row["error"] = "plugin marker; closure not compared" echoed.append(row) return { "graphSource": graph_source, "advisory": "unknown", "partial": True, "diffReliable": False, "upgrades": echoed, "summary": _empty_upgrade_summary(), "dependencies": _empty_upgrade_dependencies(), "vulnerabilities": _empty_vuln_buckets(), "targets": [], "notes": ["Plugin marker; closure not compared."], } def _node_errors_from_depsdev(nodes: List[Dict[str, Any]]) -> List[Dict[str, Any]]: out = [] for node in nodes: raw = node.get("errors") or [] if not isinstance(raw, list): raw = [raw] errors = [str(item) for item in raw if item] if not errors: continue out.append({ "gav": { "groupId": node.get("groupId") or "", "artifactId": node.get("artifactId") or "", "version": node.get("version") or "", }, "errors": errors, }) return out def _side_from_depsdev_graph(fetched: Dict[str, Any]) -> Dict[str, Any]: """Copy a GetDependencies payload onto the shared side shape. ``ok`` stays true when the graph is truncated or carries ``graphError``; those bits only clear ``diffReliable``. A capability flag is a hard failure. """ raw_cap = fetched.get("capabilityUnavailable") capability = raw_cap if isinstance(raw_cap, str) and raw_cap else None ok = bool(fetched.get("ok")) and capability is None truncated = bool(fetched.get("truncated")) raw_graph_error = fetched.get("graphError") graph_error = raw_graph_error.strip() if isinstance(raw_graph_error, str) else None if graph_error == "": graph_error = None nodes = [node for node in (fetched.get("nodes") or []) if isinstance(node, dict)] node_errors = _node_errors_from_depsdev(nodes) trees: List[Dict[str, Any]] = [] graph: Dict[str, Any] = {"nodes": [], "edges": []} if ok: # Index and non-int edges are already handled by _normalise_closure_graph. trees = [{ "nodes": fetched.get("nodes") or [], "edges": fetched.get("edges") or [], }] graph = _normalise_closure_graph(trees) raw_error = fetched.get("error") error = None if not ok: error = raw_error if isinstance(raw_error, str) and raw_error else "deps.dev graph unavailable" return { "ok": ok, "truncated": truncated, "diffReliable": ok and not truncated and graph_error is None and not node_errors, "error": error, "graphError": graph_error, "nodeErrors": node_errors, "capabilityUnavailable": capability, "exitCode": None, "trees": trees, "graph": graph, } def _fetch_depsdev_closure_sides( upgrade: Dict[str, str], ) -> Tuple[Dict[str, Any], Dict[str, Any]]: versions = [upgrade["fromVersion"], upgrade["toVersion"]] def fetch(version: str) -> Dict[str, Any]: try: return fetch_depsdev_dependencies( upgrade["groupId"], upgrade["artifactId"], version, ) except Exception: # Do not interpolate exception text: urlopen errors can embed userinfo. return { "ok": False, "error": "deps.dev graph fetch failed", "capabilityUnavailable": "unreachable", "nodes": [], "edges": [], "truncated": False, "graphError": None, "partial": True, } rows, _partial = _map_parallel(versions, fetch, deadline=_now() + TOOL_DEADLINE) sides: List[Dict[str, Any]] = [] for row in rows: if not isinstance(row, dict): row = { "ok": False, "error": "deps.dev graph fetch did not finish before the deadline", "capabilityUnavailable": "unreachable", "nodes": [], "edges": [], "truncated": False, "graphError": None, } sides.append(_side_from_depsdev_graph(row)) return sides[0], sides[1] def _license_row_for_gav(gav: Dict[str, str], row: Optional[Dict[str, Any]]) -> Dict[str, Any]: base: Dict[str, Any] = { "groupId": gav["groupId"], "artifactId": gav["artifactId"], "version": gav["version"], "ok": False, "licenses": [], } if row is None: base["error"] = "deps.dev license fetch did not finish before the deadline" base["capabilityUnavailable"] = "unreachable" return base base["ok"] = bool(row.get("ok")) base["licenses"] = list(row.get("licenses") or []) if row.get("status") is not None: base["status"] = row["status"] raw_error = row.get("error") if raw_error: base["error"] = raw_error if isinstance(raw_error, str) else "deps.dev license fetch failed" if row.get("capabilityUnavailable"): base["capabilityUnavailable"] = row["capabilityUnavailable"] elif not base["ok"] and row.get("status") != 404: # 404 is a missing deps.dev record. 5xx, bad JSON, and a deadline # miss are a coverage gap the ranker already treats as review. base["capabilityUnavailable"] = "unreachable" return base def _fetch_one_upgrade_license(gav: Dict[str, str]) -> Dict[str, Any]: try: row = fetch_depsdev_licenses(gav["groupId"], gav["artifactId"], gav["version"]) except Exception: row = { "ok": False, "licenses": [], "error": "deps.dev license fetch failed", "capabilityUnavailable": "unreachable", } return _license_row_for_gav(gav, row if isinstance(row, dict) else None) def _fetch_upgrade_license_delta( diff: Dict[str, Any], upgrades: List[Dict[str, str]], *, project_license: Optional[str], disallow: Optional[List[str]], ) -> Optional[Dict[str, Any]]: """GetVersion for the closure delta only. One GAV skips ``_map_parallel``.""" selected = _select_delta_gavs( upgrades, diff, include_targets=False, cap=MAX_UPGRADE_LICENSE_NODES, ) gavs = selected["gavs"] if not gavs: return None if len(gavs) == 1: fetched = [_fetch_one_upgrade_license(gavs[0])] else: rows, _partial = _map_parallel( gavs, _fetch_one_upgrade_license, deadline=_now() + TOOL_DEADLINE, ) fetched = [] for gav, row in zip(gavs, rows): if isinstance(row, dict): fetched.append(row) else: fetched.append(_license_row_for_gav(gav, None)) delta = _license_delta( diff, fetched, project_license=project_license, disallow=disallow, ) if selected.get("truncated"): delta["truncated"] = True return delta def _upgrade_include_licenses(args: Dict) -> bool: raw = args.get("includeLicenses", True) if raw is None: return True return bool(raw) def _upgrade_license_policy_args( args: Dict, ) -> Tuple[Optional[str], Optional[List[str]]]: project_license = args.get("projectLicense") if not isinstance(project_license, str): project_license = None disallow = args.get("disallow") if disallow is None: return project_license, None if not isinstance(disallow, list): disallow = [str(disallow)] return project_license, [str(item) for item in disallow] def _graph_failure_fields( before: Dict[str, Any], after: Dict[str, Any], ) -> Tuple[str, Optional[str]]: error = None capability = None for side in (before, after): if side.get("ok"): continue if error is None and isinstance(side.get("error"), str) and side.get("error"): error = side["error"] cap = side.get("capabilityUnavailable") if capability is None and isinstance(cap, str) and cap: capability = cap return error or "upgrade graph unavailable", capability # Static init script. Coordinates stay in the JSON file, never in this text. # @@NODE_CAP@@ is MAX_TRANSITIVE_GRAPH_NODES, not a requested coordinate, so # two phases still share one byte string. _GRADLE_UPGRADE_INIT_SCRIPT = r""" import groovy.json.JsonSlurper import org.gradle.api.artifacts.result.ResolvedDependencyResult import org.gradle.api.artifacts.result.UnresolvedDependencyResult import org.gradle.api.artifacts.component.ModuleComponentIdentifier import org.gradle.api.artifacts.component.ProjectComponentIdentifier // Duplicates _is_production_runtime_configuration and // _select_configurations_to_resolve. Do not read resolutionResult unless the // name passes: that read resolves the configuration. Test configurations and // the buildscript classpath are not resolved here. def isTestConfiguration(String name) { if (name.startsWith("test")) { return true } return name.matches(".*[a-z]Test.*") } def isProductionRuntime(String name) { if (name == null || name.isEmpty()) { return false } if (isTestConfiguration(name)) { return false } if (name == "classpath" || name == "compileOnly") { return false } if (name == "compileClasspath" || name.endsWith("CompileClasspath")) { return false } // endswith("RuntimeClasspath") misses Java's runtimeClasspath (lowercase r). return name == "runtimeClasspath" || name.endsWith("RuntimeClasspath") } def oneLine(message) { def text = message == null ? "resolution failed" : message.toString() text = text.replace("\t", " ").replace("\n", " ").replace("\r", " ") if (text.length() > 500) { text = text.substring(0, 500) } return text } def field(value) { def text = value == null ? "" : value.toString() return text.replace("\t", " ").replace("\n", " ").replace("\r", " ") } def upgradeFile = System.getProperty("maven.mcp.upgrade.file") def payload = new JsonSlurper().parse(new File(upgradeFile)) def phase = payload.phase == null ? "" : payload.phase.toString() def substitution = payload.substitution == null ? "exact" : payload.substitution.toString() def upgrades = payload.upgrades def matched = new LinkedHashMap() upgrades.each { u -> def key = u.groupId + "\t" + u.artifactId + "\t" + u.fromVersion + "\t" + u.toVersion matched[key] = [0, 0] } def requests = [] def recordRequest = { group, name, requestedVersion, rewritten -> requests.add([group, name, requestedVersion, rewritten]) } // useVersion does not add a strict constraint. Landed is decided in Python. def applyRule = { details -> def req = details.requested def requestedVersion = req.version == null ? "" : req.version.toString() upgrades.each { u -> def sameModule = req.group == u.groupId && req.name == u.artifactId if (!sameModule) { return } def exact = req.version != null && req.version.toString() == u.fromVersion.toString() def versionless = req.version == null || req.version.toString().isEmpty() def take = substitution == "module" || exact || versionless if (!take) { recordRequest(field(req.group), field(req.name), field(requestedVersion), "0") return } def key = u.groupId + "\t" + u.artifactId + "\t" + u.fromVersion + "\t" + u.toVersion def counts = matched[key] if (counts != null) { counts[0] = ((int) counts[0]) + 1 if (versionless) { counts[1] = ((int) counts[1]) + 1 } } details.useVersion(u.toVersion.toString()) details.because("maven-mcp upgrade preview") recordRequest(field(u.groupId), field(u.artifactId), field(requestedVersion), "1") } } def walkComponent walkComponent = { component, parentIndex, state -> if (state.truncated) { return } def id = component.id def key = String.valueOf(id) if (state.seen.containsKey(key)) { if (parentIndex != null) { state.edges.add([parentIndex, state.seen[key]]) } return } if (((int) state.next) >= @@NODE_CAP@@) { state.truncated = true return } def index = (int) state.next state.next = index + 1 state.seen[key] = index def projectNode = id instanceof ProjectComponentIdentifier if (projectNode || !(id instanceof ModuleComponentIdentifier)) { state.nodes.add([index: index, project: true]) } else { state.nodes.add([ index: index, group: id.group, artifact: id.module, version: id.version, ]) } if (parentIndex != null) { state.edges.add([parentIndex, index]) } def dependencies = component.dependencies if (dependencies == null) { return } dependencies.each { dep -> if (state.truncated) { return } if (dep instanceof UnresolvedDependencyResult) { def requested = dep.requested def label = requested == null ? "unresolved dependency" : requested.displayName def failure = dep.failure def message = failure == null || failure.message == null ? label : failure.message state.errors.add(oneLine(message)) return } if (dep instanceof ResolvedDependencyResult) { def again = walkComponent again.call(dep.selected, index, state) } } } def dumpConfiguration = { p, cfg -> try { def state = [ nodes: [], edges: [], seen: [:], next: 0, truncated: false, errors: [], ] // Read resolutionResult only while the help task holds the project lock. // Gradle 9 rejects the same read from projectsEvaluated. The walk is a // script closure; call it explicitly so Gradle's delegate does not // swallow the name. def walk = walkComponent walk.call(cfg.incoming.resolutionResult.root, null, state) println("===MAVEN_MCP_TREE===\t" + p.path + "\t" + cfg.name) state.nodes.each { node -> if (node.project) { println("PROJECT\t" + node.index) } else { println( "NODE\t" + node.index + "\t" + field(node.group) + "\t" + field(node.artifact) + "\t" + field(node.version) ) } } state.edges.each { edge -> println("EDGE\t" + edge[0] + "\t" + edge[1]) } println("===MAVEN_MCP_TREE_END===") if (state.truncated) { println("===MAVEN_MCP_TRUNCATED===\t" + p.path + "\t" + cfg.name) } state.errors.each { message -> println("===MAVEN_MCP_CONFIG_ERROR===\t" + p.path + "\t" + cfg.name + "\t" + message) } } catch (Exception e) { println("===MAVEN_MCP_CONFIG_ERROR===\t" + p.path + "\t" + cfg.name + "\t" + oneLine(e.message)) } } gradle.projectsEvaluated { def planned = [] gradle.rootProject.allprojects.each { p -> def selected = [] p.configurations.each { cfg -> if (cfg.canBeResolved && isProductionRuntime(cfg.name)) { selected.add(cfg) } } def hasRelease = selected.any { cfg -> cfg.name.startsWith("release") && cfg.name.endsWith("RuntimeClasspath") } if (hasRelease) { selected = selected.findAll { cfg -> cfg.name != "runtimeClasspath" } } selected.each { cfg -> if (phase == "after") { cfg.resolutionStrategy.eachDependency applyRule } planned.add([project: p, configuration: cfg]) } } def help = gradle.rootProject.tasks.findByName("help") if (help != null) { help.doFirst { def dump = dumpConfiguration planned.each { item -> dump.call(item.project, item.configuration) } matched.each { key, counts -> println("===MAVEN_MCP_MATCHED===\t" + key + "\t" + counts[0] + "\t" + counts[1]) } requests.each { row -> println("===MAVEN_MCP_REQUEST===\t" + row[0] + "\t" + row[1] + "\t" + row[2] + "\t" + row[3]) } } } } """ def _generate_gradle_upgrade_init_script() -> str: """Static Groovy. Group, artifact, and version are not interpolated.""" return _GRADLE_UPGRADE_INIT_SCRIPT.replace( "@@NODE_CAP@@", str(MAX_TRANSITIVE_GRAPH_NODES), ) def _parse_gradle_closure_stdout(stdout: str) -> Dict[str, Any]: """Tab-separated dump. The version is the remainder after a fixed split. A space-separated ``NODE`` line is not a node. Project components are marked so ``_normalise_closure_graph`` drops them and keeps their children. """ trees: List[Dict[str, Any]] = [] errors: List[Dict[str, str]] = [] truncated: List[Dict[str, str]] = [] matched: List[Dict[str, Any]] = [] requests: List[Dict[str, Any]] = [] current: Optional[Dict[str, Any]] = None def header(line: str, marker: str) -> Optional[List[str]]: if line == marker or line.startswith(marker + "\t"): return line.split("\t") return None for line in stdout.splitlines(): if line.startswith(_GRADLE_UPGRADE_TREE_END): if current is not None: trees.append(current) current = None continue tree_header = header(line, _GRADLE_UPGRADE_TREE) if tree_header is not None: if current is not None: trees.append(current) current = { "module": tree_header[1] if len(tree_header) > 1 else "", "configuration": tree_header[2] if len(tree_header) > 2 else "", "nodes": [], "edges": [], } continue if current is not None and line.startswith("NODE\t"): parts = line.split("\t", 4) if len(parts) != 5: continue try: index = int(parts[1]) except ValueError: continue current["nodes"].append({ "index": index, "groupId": parts[2], "artifactId": parts[3], "version": parts[4], }) continue if current is not None and line.startswith("PROJECT\t"): parts = line.split("\t") if len(parts) < 2: continue try: index = int(parts[1]) except ValueError: continue current["nodes"].append({"index": index, "project": True}) continue if current is not None and line.startswith("EDGE\t"): parts = line.split("\t") if len(parts) < 3: continue try: src = int(parts[1]) dst = int(parts[2]) except ValueError: continue current["edges"].append({"from": src, "to": dst}) continue truncated_header = header(line, _GRADLE_UPGRADE_TRUNCATED) if truncated_header is not None: truncated.append({ "module": truncated_header[1] if len(truncated_header) > 1 else "", "configuration": truncated_header[2] if len(truncated_header) > 2 else "", }) continue error_header = header(line, _GRADLE_UPGRADE_CONFIG_ERROR) if error_header is not None: # Message keeps the remainder so a tab inside it is not a new column. parts = line.split("\t", 3) errors.append({ "module": parts[1] if len(parts) > 1 else "", "configuration": parts[2] if len(parts) > 2 else "", "message": (parts[3] if len(parts) > 3 else "")[:_GRADLE_ERROR_LIMIT], }) continue if line.startswith(_GRADLE_UPGRADE_MATCHED + "\t"): parts = line.split("\t") if len(parts) < 7: continue try: rewritten = int(parts[5]) versionless = int(parts[6]) except ValueError: continue matched.append({ "groupId": parts[1], "artifactId": parts[2], "fromVersion": parts[3], "toVersion": parts[4], "rewritten": rewritten, "versionless": versionless, }) continue if line.startswith(_GRADLE_UPGRADE_REQUEST + "\t"): parts = line.split("\t") if len(parts) < 5: continue try: rewritten = int(parts[4]) except ValueError: continue requests.append({ "groupId": parts[1], "artifactId": parts[2], "requestedVersion": parts[3], "rewritten": rewritten, }) if current is not None: trees.append(current) return { "trees": trees, "errors": errors, "truncated": truncated, "matched": matched, "requests": requests, } def _filter_gradle_closure_parse( parsed: Dict[str, Any], ) -> Tuple[List[Dict[str, Any]], List[Dict[str, str]], bool]: """Drop trees the Python production-runtime predicate rejects.""" trees = list(parsed.get("trees") or []) errors = list(parsed.get("errors") or []) truncated_keys = { (item.get("module"), item.get("configuration")) for item in parsed.get("truncated") or [] } names: Dict[str, List[str]] = {} for tree in trees: names.setdefault(tree["module"], []).append(tree["configuration"]) for error in errors: names.setdefault(error["module"], []).append(error["configuration"]) selected: Dict[str, Set[str]] = {} for module, configs in names.items(): unique: List[str] = [] for config in configs: if config not in unique: unique.append(config) selected[module] = set(_select_configurations_to_resolve(unique)) kept_trees = [ tree for tree in trees if tree["configuration"] in selected.get(tree["module"], set()) ] kept_errors = [ error for error in errors if error["configuration"] in selected.get(error["module"], set()) ] truncated = any( (tree["module"], tree["configuration"]) in truncated_keys for tree in kept_trees ) return kept_trees, kept_errors, truncated def _failed_gradle_side(message: str, exit_code: Optional[int] = None) -> Dict[str, Any]: return { "ok": False, "truncated": False, "diffReliable": False, "error": message[:_GRADLE_ERROR_LIMIT], "graphError": None, "nodeErrors": [], "capabilityUnavailable": None, "exitCode": exit_code, "trees": [], "graph": {"nodes": [], "edges": []}, "matched": [], } def _gradle_unavailable_error( exit_code: Optional[int], stderr: str, config_error: Optional[str], ) -> str: """Graph-source failure text. Exit 0 does not copy stderr. Stderr is truncated. Exception text from Python is not interpolated. """ if exit_code == 0: if config_error: return config_error[:_GRADLE_ERROR_LIMIT] return "Gradle produced no dependency trees" text = (stderr or "").strip() if text: return text[:_GRADLE_ERROR_LIMIT] if config_error: return config_error[:_GRADLE_ERROR_LIMIT] if exit_code == 124: return "Gradle command timed out" if exit_code is None: return "Gradle was not launched" return f"Gradle exited with code {exit_code} and produced no dependency trees" def _side_from_gradle_output( exit_code: Optional[int], stdout: str, stderr: str, ) -> Dict[str, Any]: """One before or after run onto the shared side shape. Zero trees is ``ok: false``, including exit 0. A non-zero exit that still printed a production tree stays ``ok`` and clears ``diffReliable``. """ parsed = _parse_gradle_closure_stdout(stdout or "") kept_trees, kept_errors, truncated = _filter_gradle_closure_parse(parsed) for tree in kept_trees: if tree.get("module") in ("", ":"): tree["module"] = None graph_error = None if kept_errors: parts = [] for item in kept_errors: module = item.get("module") or ":" config = item.get("configuration") or "" message = item.get("message") or "resolution failed" parts.append(f"{module}:{config} {message}".strip()) graph_error = "; ".join(parts)[:_GRADLE_ERROR_LIMIT] ok = bool(kept_trees) if ok: error = None graph = _normalise_closure_graph(kept_trees) trees_out: List[Dict[str, Any]] = kept_trees else: error = _gradle_unavailable_error(exit_code, stderr, graph_error) graph = {"nodes": [], "edges": []} trees_out = [] graph_error = None diff_reliable = ok and exit_code == 0 and not truncated and not kept_errors return { "ok": ok, "truncated": truncated, "diffReliable": diff_reliable, "error": error, "graphError": graph_error, "nodeErrors": [], "capabilityUnavailable": None, "exitCode": exit_code, "trees": trees_out, "graph": graph, "matched": list(parsed.get("matched") or []), "requests": list(parsed.get("requests") or []), } def _gradle_resolution_fields( upgrade: Dict[str, str], after_side: Dict[str, Any], ) -> Dict[str, Any]: """Landed means every selected version on dumped configurations is ``toVersion``. ``substitutionMatched > 0`` is not landed. An absent GA is not landed. """ group_id = upgrade["groupId"] artifact_id = upgrade["artifactId"] to_version = upgrade["toVersion"] selected: List[str] = [] for node in (after_side.get("graph") or {}).get("nodes") or []: if node.get("groupId") != group_id or node.get("artifactId") != artifact_id: continue version = node.get("version") if version and version not in selected: selected.append(version) selected = _sorted_versions(selected) rewritten = 0 versionless = 0 for row in after_side.get("matched") or []: if ( row.get("groupId") == group_id and row.get("artifactId") == artifact_id and row.get("fromVersion") == upgrade["fromVersion"] and row.get("toVersion") == to_version ): rewritten = int(row.get("rewritten") or 0) versionless = int(row.get("versionless") or 0) break return { "selectedVersions": selected, "targetPresent": bool(selected), "landed": bool(selected) and all(version == to_version for version in selected), "substitutionMatched": rewritten, "rewroteVersionless": versionless > 0, } def _gradle_closure_trees( project_path: str, upgrades: List[Dict[str, str]], substitution: str, ) -> Dict[str, Any]: """Two sequential wrapper runs. Missing ``gradlew`` does not launch. The init script file is written once. ``phase`` changes only in the JSON file. The temp directory is removed when this returns. """ gradlew = _find_gradle_wrapper(project_path) if not gradlew: failed = _failed_gradle_side("Gradle wrapper (gradlew) not found") return { "launched": False, "before": failed, "after": failed, "requests": [], } with tempfile.TemporaryDirectory(prefix="maven-mcp-upgrade-") as tmp: script_path = os.path.join(tmp, "maven-mcp-upgrade.init.gradle") json_path = os.path.join(tmp, "upgrades.json") script = _generate_gradle_upgrade_init_script() with open(script_path, "w", encoding="utf-8") as handle: handle.write(script) def run(phase: str) -> Tuple[int, str, str]: payload = { "phase": phase, "substitution": substitution, "upgrades": [ { "groupId": upgrade["groupId"], "artifactId": upgrade["artifactId"], "fromVersion": upgrade["fromVersion"], "toVersion": upgrade["toVersion"], } for upgrade in upgrades ], } with open(json_path, "w", encoding="utf-8") as handle: json.dump(payload, handle) # --offline, when set, is prepended by _gradle_cli_prefix_args. args = [ "--no-configuration-cache", "--init-script", script_path, "-q", "help", "-Dmaven.mcp.upgrade.file=" + json_path, ] return _run_gradle_command( project_path, gradlew, args, timeout=_gradle_resolve_timeout_seconds(), ) before_code, before_out, before_err = run("before") after_code, after_out, after_err = run("after") before = _side_from_gradle_output(before_code, before_out, before_err) after = _side_from_gradle_output(after_code, after_out, after_err) return { "launched": True, "before": before, "after": after, "requests": list(after.get("requests") or []), } def _partition_upgrade_requests( upgrades: List[Dict[str, str]], ) -> Tuple[List[Dict[str, Any]], List[Dict[str, str]]]: """Marker rows carry an error. Libraries are what a resolve may see.""" echoed: List[Dict[str, Any]] = [] libraries: List[Dict[str, str]] = [] for upgrade in upgrades: row = dict(upgrade) if _gradle_plugin_marker_plugin_id(upgrade["groupId"], upgrade["artifactId"]): row["error"] = "plugin marker; closure not compared" else: libraries.append(upgrade) echoed.append(row) return echoed, libraries def compare_upgrade_closure(args: Dict) -> Dict[str, Any]: """Compare a direct upgrade before and after the candidate version. ``auto`` uses Gradle when a wrapper exists, otherwise one deps.dev pair. A Gradle failure is not replaced with deps.dev. ``advisory`` is not a safety verdict. ``get_transitive_graph`` is not used (it drops ``relation``). """ project_path = _upgrade_project_path(args) graph_source = _select_upgrade_graph_source(args, project_path) upgrades = _parse_upgrade_requests(args.get("upgrades")) if graph_source == "depsdev" and len(upgrades) != 1: raise ValueError("deps.dev compares exactly one upgrade") substitution = _upgrade_substitution(args) upgrades_out, libraries = _partition_upgrade_requests(upgrades) if not libraries: result = _marker_upgrade_result(graph_source, upgrades_out) _log_upgrade_closure(result) return result if all(upgrade["fromVersion"] == upgrade["toVersion"] for upgrade in libraries): ranked = _advisory_for_upgrade(all_identity=True, targets=libraries) result = _with_upgrade_source(graph_source, upgrades_out, ranked) if any(upgrade.get("error") for upgrade in upgrades_out): # A marker in an otherwise unchanged batch is not "none". result["advisory"] = "unknown" result["partial"] = True result["targets"] = [] result["notes"] = [ "Plugin marker; closure not compared.", _ADVISORY_IDENTITY_NOTE, ] _log_upgrade_closure(result) return result include_licenses = _upgrade_include_licenses(args) project_license, disallow = _upgrade_license_policy_args(args) launched_gradle = False requests: Optional[List[Dict[str, Any]]] = None if graph_source == "gradle": fetched = _gradle_closure_trees(project_path, libraries, substitution) before = fetched["before"] after = fetched["after"] requests = list(fetched.get("requests") or []) launched_gradle = bool(fetched.get("launched")) else: before, after = _fetch_depsdev_closure_sides(libraries[0]) if not before["ok"] or not after["ok"]: ranked = _advisory_for_upgrade( targets=[dict(upgrade) for upgrade in libraries], before_ok=bool(before["ok"]), after_ok=bool(after["ok"]), ) error, capability = _graph_failure_fields(before, after) result = _with_upgrade_source( graph_source, upgrades_out, ranked, error=error, capability=capability, ) if requests is not None: result["requests"] = requests _log_upgrade_closure( result, before=before, after=after, launched_gradle=launched_gradle, ) return result compared = libraries if graph_source == "gradle": compared = [] for upgrade in libraries: row = dict(upgrade) row.update(_gradle_resolution_fields(upgrade, after)) compared.append(row) diff = _diff_closure( before, after, compared, project_graph=(graph_source == "gradle"), ) selected = _select_delta_gavs(compared, diff) records = query_osv_batch(selected["gavs"]) if selected["gavs"] else [] classified = _classify_vuln_delta(compared, diff, records) license_delta = None if include_licenses: license_delta = _fetch_upgrade_license_delta( diff, compared, project_license=project_license, disallow=disallow, ) if graph_source == "gradle": notes = _gradle_closure_notes(include_licenses) else: notes = _depsdev_closure_notes(include_licenses) if any(row.get("rewroteVersionless") for row in compared): notes = list(notes) + [_VERSIONLESS_ARM_NOTE] ranked = _advisory_for_upgrade( targets=classified["targets"], before_ok=True, after_ok=True, diff_reliable=bool(diff.get("diffReliable")), diff=diff, vulnerabilities=classified["vulnerabilities"], license_delta=license_delta, fixes_incomplete=bool(classified.get("fixesIncomplete")), capability_unavailable=classified.get("capabilityUnavailable"), input_truncated=bool(classified.get("inputTruncated") or selected.get("truncated")), ) if any(row.get("error") for row in upgrades_out) and ranked.get("advisory") == "none": ranked = dict(ranked) ranked["advisory"] = "review" ranked["partial"] = True ranked["notes"] = [ "A requested coordinate was not closure-checked.", *list(ranked.get("notes") or []), ] result = _with_upgrade_source( graph_source, upgrades_out, ranked, truncated=bool(before.get("truncated") or after.get("truncated")), notes=notes, ) if requests is not None: result["requests"] = requests _log_upgrade_closure( result, before=before, after=after, launched_gradle=launched_gradle, ) return result def _detect_conflicts_from_gradle_scan(scan: Dict[str, Any]) -> Dict[str, Any]: """Detect version conflicts from Gradle-resolved scan usages (highest-wins).""" strategy = strategy_for_build_system("gradle") bs_label = scan.get("buildSystem") or "gradle" # ga -> version -> {sources: [usage labels…]} ga_versions: Dict[Tuple[str, str], Dict[str, Dict[str, Any]]] = {} for dep in scan.get("dependencies") or []: if dep.get("isPlatform"): continue gid, aid = dep.get("groupId"), dep.get("artifactId") ver = dep.get("version") or dep.get("effectiveVersion") if not gid or not aid or not ver: continue usages = dep.get("usages") or [{"module": None, "configuration": ""}] for usage in usages: module = usage.get("module") config = usage.get("configuration", "") label = f"{module}:{config}" if module else (config or "gradle-resolved") bucket = ga_versions.setdefault((gid, aid), {}) info = bucket.setdefault(ver, {"sources": []}) if label not in info["sources"]: info["sources"].append(label) conflicts: List[Dict[str, Any]] = [] for (gid, aid), ver_map in sorted(ga_versions.items()): versions = list(ver_map.keys()) if len(versions) < 2: continue resolved = resolve_conflict_version(versions, strategy) sources: List[Dict[str, Any]] = [] for v in sorted(versions, key=functools.cmp_to_key(compare_versions)): sources.append({ "version": v, "via": ver_map[v]["sources"][:10], "minDepth": None, }) conflicts.append({ "groupId": gid, "artifactId": aid, "versions": sorted(versions, key=functools.cmp_to_key(compare_versions)), "resolvedTo": resolved, "strategy": strategy, "risk": _conflict_risk(versions, resolved), "sources": sources, }) risk_order = {"high": 0, "medium": 1, "low": 2} conflicts.sort( key=lambda c: ( risk_order.get(c["risk"], 9), c["groupId"], c["artifactId"], ) ) notes = [ "Conflict detection derived from Gradle-resolved dependency scan usages " + "(highest-wins mediation).", "Compares versions of the same groupId:artifactId seen across module/" + "configuration usages in the resolved tree.", "Does not model ResolutionStrategy, strict versions, or enforcedPlatform " + "overrides beyond what Gradle already reported.", ] return { "buildSystem": bs_label, "strategy": strategy, "conflicts": conflicts, "scannedRoots": 0, "graphsFetched": 0, "graphsFailed": 0, "partial": False, "errors": [], "notes": notes, } def detect_dependency_conflicts( project_path: str, build_system: Optional[str] = None, ctx: Optional["ResolutionContext"] = None, ) -> Dict[str, Any]: """Detect GAs resolved at ≥2 versions across direct deps' transitive graphs. For each versioned direct dependency (capped at ``MAX_CONFLICT_SCAN_ROOTS``), fetches the deps.dev graph and unions every ``g:a → {versions…}`` seen. When a GA appears at multiple versions, reports the conflict with the version the active build system's mediation strategy would pick. This is an approximation of a full project resolve: deps.dev resolves each root in isolation (no project-wide dependencyManagement / resolutionStrategy / enforcedPlatform). Documented in ``notes``. """ if ctx is None: ctx = build_resolution_context({"projectPath": project_path}) scan = scan_project(project_path) if scan.get("resolvedBy") == "gradle": return _detect_conflicts_from_gradle_scan(scan) apply_bom_managed_versions(scan, ctx) detected_bs = scan.get("buildSystem") or "unknown" bs = (build_system or detected_bs or "unknown").lower() if bs not in ("maven", "gradle"): # unknown → treat as gradle highest-wins (documented). strategy = strategy_for_build_system(bs) bs_label = detected_bs else: strategy = strategy_for_build_system(bs) bs_label = bs def _eff(dep: Dict) -> Optional[str]: return dep.get("version") or dep.get("effectiveVersion") roots: List[Dict] = [] for dep in scan.get("dependencies") or []: if dep.get("isPlatform"): continue ver = _eff(dep) gid, aid = dep.get("groupId"), dep.get("artifactId") if not gid or not aid or not ver: continue roots.append({"groupId": gid, "artifactId": aid, "version": ver}) # Dedupe identical GAV roots (same direct declared twice across modules). seen_roots: set = set() unique_roots: List[Dict] = [] for r in roots: key = (r["groupId"], r["artifactId"], r["version"]) if key in seen_roots: continue seen_roots.add(key) unique_roots.append(r) truncated_roots = len(unique_roots) > MAX_CONFLICT_SCAN_ROOTS unique_roots = unique_roots[:MAX_CONFLICT_SCAN_ROOTS] # ga -> { version -> {sources: [root gav…], minDepth: int|None } } ga_versions: Dict[Tuple[str, str], Dict[str, Dict[str, Any]]] = {} errors: List[str] = [] graphs_ok = 0 graphs_failed = 0 capability: Optional[str] = None # #400: fetch every root's transitive graph in parallel (pure, independent # per-root network calls; no shared state touched) — then aggregate # SEQUENTIALLY below in the ORIGINAL root order, exactly as before. Unlike # check_license_compliance's per-node walk, this aggregation (min-depth, # append-to-sources) has no order-sensitive "upgrade in place" logic, so # keeping it sequential is purely for a minimal/obviously-correct diff, # not a correctness requirement. def _fetch_conflict_root_graph(root: Dict[str, str]) -> Dict[str, Any]: return fetch_depsdev_dependencies( root["groupId"], root["artifactId"], root["version"] ) root_graphs, _partial = _map_parallel( unique_roots, _fetch_conflict_root_graph, max_workers=MAX_PARALLEL_FETCHES, ) for root, fetched in zip(unique_roots, root_graphs): root_label = f"{root['groupId']}:{root['artifactId']}:{root['version']}" if fetched.get("capabilityUnavailable") and capability is None: capability = fetched["capabilityUnavailable"] if not fetched["ok"]: graphs_failed += 1 if len(errors) < MAX_DEPSDEV_ERRORS_REPORTED: errors.append(f"{root_label}: {fetched.get('error') or 'unavailable'}") continue graphs_ok += 1 if fetched.get("graphError") and len(errors) < MAX_DEPSDEV_ERRORS_REPORTED: errors.append(f"{root_label}: graph error: {fetched['graphError']}") depths = _edge_depth_map(fetched["edges"], 0) for idx, node in enumerate(fetched["nodes"]): ga = (node["groupId"], node["artifactId"]) ver = node["version"] if not ver: continue bucket = ga_versions.setdefault(ga, {}) info = bucket.setdefault(ver, {"sources": [], "minDepth": None}) if root_label not in info["sources"]: info["sources"].append(root_label) d = depths.get(idx) if d is not None and (info["minDepth"] is None or d < info["minDepth"]): info["minDepth"] = d conflicts: List[Dict[str, Any]] = [] for (gid, aid), ver_map in sorted(ga_versions.items()): versions = list(ver_map.keys()) if len(versions) < 2: continue depths_by_version = { v: ver_map[v]["minDepth"] for v in versions if ver_map[v]["minDepth"] is not None } resolved = resolve_conflict_version( versions, strategy, depths_by_version=depths_by_version ) sources: List[Dict[str, Any]] = [] for v in sorted(versions, key=functools.cmp_to_key(compare_versions)): sources.append({ "version": v, "via": ver_map[v]["sources"][:10], "minDepth": ver_map[v]["minDepth"], }) conflicts.append({ "groupId": gid, "artifactId": aid, "versions": sorted(versions, key=functools.cmp_to_key(compare_versions)), "resolvedTo": resolved, "strategy": strategy, "risk": _conflict_risk(versions, resolved), "sources": sources, }) # Sort conflicts by risk then GA for stable output. risk_order = {"high": 0, "medium": 1, "low": 2} conflicts.sort( key=lambda c: ( risk_order.get(c["risk"], 9), c["groupId"], c["artifactId"], ) ) notes = [ "Conflict detection unions deps.dev graphs for each direct dependency " + "resolved in isolation — not a full project-wide Maven/Gradle resolve.", "Maven nearest-wins uses BFS depth from each direct root; declaration-" + "order tie-breaks inside the same depth are approximated by highest version.", "Gradle highest-wins ignores project ResolutionStrategy / strict versions / " + "enforcedPlatform overrides.", "Private/unpublished coordinates and deps.dev coverage gaps degrade per-root " + "(see errors[]); remaining roots still contribute.", ] if truncated_roots: notes.append( f"Direct roots truncated to {MAX_CONFLICT_SCAN_ROOTS} " + "(MAX_CONFLICT_SCAN_ROOTS); results are partial." ) out = { "buildSystem": bs_label, "strategy": strategy, "conflicts": conflicts, "scannedRoots": len(unique_roots), "graphsFetched": graphs_ok, "graphsFailed": graphs_failed, "partial": bool(truncated_roots or graphs_failed or errors), "errors": errors, "notes": notes, } return _with_capability(out, capability) # --------------------------------------------------------------------------- # Version compatibility matrices (#285) # --------------------------------------------------------------------------- SPRING_BOOT_BOM_GROUP = "org.springframework.boot" SPRING_BOOT_BOM_ARTIFACT = "spring-boot-dependencies" COMPAT_MATRICES_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "compat-matrices.json") MAX_COMPAT_DEPENDENCIES = 100 _COMPAT_MATRICES_CACHE: Optional[Dict] = None def _load_compat_matrices() -> Dict: """Load shipped compat-matrices.json (cached for the process lifetime).""" global _COMPAT_MATRICES_CACHE if _COMPAT_MATRICES_CACHE is not None: return _COMPAT_MATRICES_CACHE with open(COMPAT_MATRICES_PATH, "r", encoding="utf-8") as fh: _COMPAT_MATRICES_CACHE = json.load(fh) return _COMPAT_MATRICES_CACHE def _major_minor_line(version: str) -> str: """Return ``major.minor`` for matrix line matching (e.g. 8.7.3 → 8.7).""" segs = _parse_segments(version) if len(segs) >= 2: return f"{segs[0]}.{segs[1]}" if segs: return str(segs[0]) return version def _version_gte(a: str, b: str) -> bool: return compare_versions(a, b) >= 0 def _version_lte(a: str, b: str) -> bool: return compare_versions(a, b) <= 0 def _version_in_closed_range(version: str, vmin: str, vmax: str) -> bool: return _version_gte(version, vmin) and _version_lte(version, vmax) def _find_agp_entry(agp_version: str, entries: List[Dict]) -> Optional[Dict]: line = _major_minor_line(agp_version) for entry in entries: if entry.get("agp") == line: return entry for entry in entries: if entry.get("agp") == agp_version: return entry return None def _find_kgp_entry(kotlin_version: str, entries: List[Dict]) -> Optional[Dict]: for entry in entries: kmin = entry.get("kgpMin") or "" kmax = entry.get("kgpMax") or "" if kmin and kmax and _version_in_closed_range(kotlin_version, kmin, kmax): return entry return None def _nearest_agp_suggestion(agp_version: str, entries: List[Dict]) -> Optional[Dict]: """Pick the closest AGP matrix row by major.minor distance for suggestions.""" if not entries: return None target = _parse_segments(agp_version) best = None best_dist = None for entry in entries: segs = _parse_segments(entry["agp"]) # Lexicographic distance on padded major.minor a = (target + [0, 0])[:2] b = (segs + [0, 0])[:2] dist = abs(a[0] - b[0]) * 1000 + abs(a[1] - b[1]) if best_dist is None or dist < best_dist: best_dist = dist best = entry return best def check_android_kotlin_compatibility(android: Dict, matrices: Dict) -> Tuple[List[Dict], List[str]]: """Validate AGP/Gradle/Kotlin/JDK against shipped matrices. Returns (conflicts, notes). """ conflicts: List[Dict] = [] notes: List[str] = [] agp = (android.get("agp") or "").strip() or None gradle = (android.get("gradle") or "").strip() or None kotlin = (android.get("kotlin") or "").strip() or None jdk_raw = android.get("jdk") jdk: Optional[int] = None if jdk_raw is not None and jdk_raw != "": try: jdk = int(str(jdk_raw).strip()) except (TypeError, ValueError): notes.append(f"android.jdk={jdk_raw!r} is not an integer; JDK check skipped.") agp_entries = matrices.get("agpEntries") or [] kgp_entries = matrices.get("kotlinGradlePluginEntries") or [] agp_entry = _find_agp_entry(agp, agp_entries) if agp else None if agp and agp_entry is None: nearest = _nearest_agp_suggestion(agp, agp_entries) suggestion = None if nearest: suggestion = { "agp": nearest["agp"], "gradle": nearest["minGradle"], "jdk": nearest["minJdk"], } conflicts.append({ "kind": "agp_unknown", "requested": {"agp": agp, "gradle": gradle, "kotlin": kotlin, "jdk": jdk}, "expected": None, "suggestion": suggestion, "reference": (nearest or {}).get("referenceUrl") or "https://developer.android.com/build/releases/about-agp", }) elif agp_entry is not None: ref = agp_entry.get("referenceUrl") or "https://developer.android.com/build/releases/about-agp" suggestion = { "agp": agp_entry["agp"], "gradle": agp_entry["minGradle"], "jdk": agp_entry["minJdk"], } if gradle and not _version_gte(gradle, agp_entry["minGradle"]): conflicts.append({ "kind": "agp_gradle", "requested": {"agp": agp, "gradle": gradle}, "expected": {"minGradle": agp_entry["minGradle"]}, "suggestion": suggestion, "reference": ref, }) if jdk is not None and jdk < int(agp_entry["minJdk"]): conflicts.append({ "kind": "agp_jdk", "requested": {"agp": agp, "jdk": jdk}, "expected": {"minJdk": agp_entry["minJdk"]}, "suggestion": suggestion, "reference": ref, }) if kotlin: kgp_entry = _find_kgp_entry(kotlin, kgp_entries) if kgp_entry is None: # Suggest the nearest band by kgpMin distance. nearest_k = None best_dist = None for entry in kgp_entries: a = _parse_segments(kotlin) b = _parse_segments(entry["kgpMin"]) pad = max(len(a), len(b), 3) a = (a + [0] * pad)[:pad] b = (b + [0] * pad)[:pad] d = sum(abs(x - y) for x, y in zip(a, b)) if best_dist is None or d < best_dist: best_dist = d nearest_k = entry suggestion = None if nearest_k: suggestion = { "kotlin": f"{nearest_k['kgpMin']}-{nearest_k['kgpMax']}", "gradle": f"{nearest_k['gradleMin']}-{nearest_k['gradleMax']}", "agp": f"{nearest_k['agpMin']}-{nearest_k['agpMax']}", } conflicts.append({ "kind": "kotlin_unknown", "requested": {"kotlin": kotlin, "gradle": gradle, "agp": agp}, "expected": None, "suggestion": suggestion, "reference": (nearest_k or {}).get("referenceUrl") or "https://kotlinlang.org/docs/gradle-configure-project.html", }) else: ref = kgp_entry.get("referenceUrl") or ( "https://kotlinlang.org/docs/gradle-configure-project.html" ) suggestion = { "kotlin": f"{kgp_entry['kgpMin']}-{kgp_entry['kgpMax']}", "gradle": f"{kgp_entry['gradleMin']}-{kgp_entry['gradleMax']}", "agp": f"{kgp_entry['agpMin']}-{kgp_entry['agpMax']}", } if gradle and not _version_in_closed_range( gradle, kgp_entry["gradleMin"], kgp_entry["gradleMax"] ): conflicts.append({ "kind": "kotlin_gradle", "requested": {"kotlin": kotlin, "gradle": gradle}, "expected": { "gradleMin": kgp_entry["gradleMin"], "gradleMax": kgp_entry["gradleMax"], }, "suggestion": suggestion, "reference": ref, }) if agp and not _version_in_closed_range( agp, kgp_entry["agpMin"], kgp_entry["agpMax"] ): conflicts.append({ "kind": "kotlin_agp", "requested": {"kotlin": kotlin, "agp": agp}, "expected": { "agpMin": kgp_entry["agpMin"], "agpMax": kgp_entry["agpMax"], }, "suggestion": suggestion, "reference": ref, }) if not agp and not kotlin: notes.append( "android block provided without agp or kotlin; nothing to check against the matrix." ) return conflicts, notes def check_spring_boot_bom_compatibility( spring_boot: str, dependencies: List[Dict], ctx: "ResolutionContext", ) -> Tuple[List[Dict], List[str]]: """Compare requested dependency versions against spring-boot-dependencies BOM.""" conflicts: List[Dict] = [] notes: List[str] = [] managed = expand_bom( SPRING_BOOT_BOM_GROUP, SPRING_BOOT_BOM_ARTIFACT, spring_boot, ctx ) if not managed: notes.append( f"Could not expand {SPRING_BOOT_BOM_GROUP}:{SPRING_BOOT_BOM_ARTIFACT}:{spring_boot}; " "Spring Boot BOM checks skipped." ) return conflicts, notes managed_map = { (m["groupId"], m["artifactId"]): m["version"] for m in managed } ref = ( "https://docs.spring.io/spring-boot/docs/current/reference/html/" "dependency-versions.html" ) for dep in dependencies: gid = dep.get("groupId") aid = dep.get("artifactId") requested = dep.get("version") if not gid or not aid or not requested: continue expected = managed_map.get((gid, aid)) if expected is None: continue if requested == expected: continue conflicts.append({ "kind": "spring_boot_bom", "requested": { "groupId": gid, "artifactId": aid, "version": requested, }, "expected": { "groupId": gid, "artifactId": aid, "version": expected, "managedBy": { "groupId": SPRING_BOOT_BOM_GROUP, "artifactId": SPRING_BOOT_BOM_ARTIFACT, "version": spring_boot, }, }, "suggestion": { "groupId": gid, "artifactId": aid, "version": expected, }, "reference": ref, }) return conflicts, notes def check_javax_jakarta_migration( spring_boot: Optional[str], dependencies: List[Dict], matrices: Dict, ) -> Tuple[List[Dict], List[str]]: """Flag javax.* EE coordinates when Spring Boot ≥ 3.""" conflicts: List[Dict] = [] notes: List[str] = [] if not spring_boot: return conflicts, notes # Major-version gate: Boot 3 milestones/RCs (3.0.0-M1) compare < 3.0.0 # under compare_versions but already require jakarta.*. boot_major = (_parse_segments(spring_boot) or [0])[0] if boot_major < 3: return conflicts, notes jmap = matrices.get("jakartaMap") or {} ref = ( "https://github.com/spring-projects/spring-boot/wiki/" "Spring-Boot-3.0-Migration-Guide" ) for dep in dependencies: gid = dep.get("groupId") or "" aid = dep.get("artifactId") or "" key = f"{gid}:{aid}" replacement = jmap.get(key) if replacement: jg, ja = replacement.split(":", 1) conflicts.append({ "kind": "javax_to_jakarta", "requested": { "groupId": gid, "artifactId": aid, "version": dep.get("version"), }, "expected": {"groupId": jg, "artifactId": ja}, "suggestion": {"groupId": jg, "artifactId": ja}, "reference": ref, }) continue # Unmapped javax.* still flagged (no concrete replacement known). if gid.startswith("javax."): conflicts.append({ "kind": "javax_to_jakarta", "requested": { "groupId": gid, "artifactId": aid, "version": dep.get("version"), }, "expected": None, "suggestion": None, "reference": ref, }) return conflicts, notes def check_version_compatibility( *, spring_boot: Optional[str] = None, android: Optional[Dict] = None, dependencies: Optional[List[Dict]] = None, ctx: Optional["ResolutionContext"] = None, ) -> Dict: """Core compatibility check used by the MCP tool (#285).""" matrices = _load_compat_matrices() conflicts: List[Dict] = [] notes: List[str] = list((matrices.get("_meta") or {}).get("limitations") or []) deps = list(dependencies or []) if len(deps) > MAX_COMPAT_DEPENDENCIES: deps = deps[:MAX_COMPAT_DEPENDENCIES] notes.append( f"dependencies truncated to {MAX_COMPAT_DEPENDENCIES} items " "(handler-enforced cap)." ) if android: c, n = check_android_kotlin_compatibility(android, matrices) conflicts.extend(c) notes.extend(n) if spring_boot: if ctx is None: notes.append( "No resolution context; Spring Boot BOM expansion skipped." ) else: c, n = check_spring_boot_bom_compatibility(spring_boot, deps, ctx) conflicts.extend(c) notes.extend(n) c, n = check_javax_jakarta_migration(spring_boot, deps, matrices) conflicts.extend(c) notes.extend(n) elif deps and any( (d.get("groupId") or "").startswith("javax.") for d in deps ): notes.append( "javax.* coordinates present but springBoot was not set; " "javax→jakarta migration check skipped (requires Spring Boot ≥ 3)." ) if not spring_boot and not android and not deps: notes.append( "No springBoot, android, or dependencies provided; nothing to check." ) # De-dupe identical limitation notes that we always attach — keep order, # but drop exact duplicate strings from the dynamic notes path. seen_notes = set() unique_notes: List[str] = [] for note in notes: if note in seen_notes: continue seen_notes.add(note) unique_notes.append(note) return { "compatible": len(conflicts) == 0, "conflicts": conflicts, "notes": unique_notes, } # --------------------------------------------------------------------------- # Version classification & comparison # --------------------------------------------------------------------------- def classify_version(version: str) -> str: for pattern, stability in STABILITY_PATTERNS: if pattern.search(version): return stability return "stable" # Digits attached to known prerelease tokens only — free-text qualifier digits # (e.g. "0.6.x-compat") must not be treated as ordinal prerelease numbers (#331). _PRERELEASE_ORDINAL_RE = re.compile( r"(?:alpha|a(?=\d)|beta|b(?=\d)|milestone|m(?=\d)|rc|cr|snapshot)[-.]?(\d+)", re.IGNORECASE, ) def _split_core_and_qualifier(version: str) -> Tuple[str, str]: """Split a version into numeric core and qualifier suffix. Qualifier begins at the earliest of: ``-`` / ``+``, a ``.`` whose next segment is non-numeric (``.Final`` / ``.RELEASE``), or a non-digit glued onto the core (``1.0.0legacy`` / ``1.0.0_legacy``). Pure numeric dotted tails (``1.0.0.1``) stay in the core. """ if not version: return "", "" i = 0 n = len(version) while i < n and version[i].isdigit(): i += 1 while i < n and version[i] == ".": j = i + 1 while j < n and version[j].isdigit(): j += 1 if j == i + 1: # Non-numeric segment after '.' — qualifier boundary (#331). return version[:i], version[i:] i = j if i < n: return version[:i], version[i:] return version, "" def _parse_segments(version: str) -> List[int]: core, _qual = _split_core_and_qualifier(version) parts = [] for p in core.split("."): if not p: continue try: parts.append(int(p)) except ValueError: parts.append(0) return parts or [0] def _extract_prerelease_numbers(version: str) -> List[int]: _core, qual = _split_core_and_qualifier(version) if not qual: return [] return [int(m.group(1)) for m in _PRERELEASE_ORDINAL_RE.finditer(qual)] def _compare_int_lists(a: List[int], b: List[int]) -> int: max_len = max(len(a), len(b), 1) for i in range(max_len): ai = a[i] if i < len(a) else 0 bi = b[i] if i < len(b) else 0 if ai != bi: return -1 if ai < bi else 1 return 0 def compare_versions(a: str, b: str) -> int: """Returns negative if a < b, positive if a > b, 0 if equal.""" core_diff = _compare_int_lists(_parse_segments(a), _parse_segments(b)) if core_diff != 0: return core_diff weight_diff = PRERELEASE_WEIGHT.get(classify_version(a), 5) - PRERELEASE_WEIGHT.get(classify_version(b), 5) if weight_diff != 0: return weight_diff has_suffix_a = bool(_split_core_and_qualifier(a)[1]) has_suffix_b = bool(_split_core_and_qualifier(b)[1]) if has_suffix_a != has_suffix_b: # #325/#331: same core, same stability class, but only one side carries # a qualifier at all — the bare version always ranks higher. return 1 if not has_suffix_a else -1 tail_diff = _compare_int_lists(_extract_prerelease_numbers(a), _extract_prerelease_numbers(b)) if tail_diff != 0: return tail_diff return -1 if a < b else (1 if a > b else 0) def find_latest_version(versions: List[str], filter_mode: str = "PREFER_STABLE") -> Optional[str]: if not versions: return None if filter_mode == "ALL": return versions[-1] # Scan from the end for last stable stable = None for v in reversed(versions): if classify_version(v) == "stable": stable = v break if filter_mode == "STABLE_ONLY": return stable # PREFER_STABLE: stable if found, otherwise last version return stable if stable is not None else versions[-1] def find_latest_version_for_current(versions: List[str], current_version: str) -> Optional[str]: """Highest version newer-or-equal to current at acceptable stability, else None. A candidate qualifies iff it is newer-than-or-equal to current AND ranks at least as stable on snapshot < alpha < beta < milestone < rc < stable; an up-to-date dependency thus returns current itself (upgradeType "none"), never a less-stable SNAPSHOT/RC as an upgrade from a more stable current (#312). None only when nothing at acceptable stability is >= current. """ current_rank = PRERELEASE_WEIGHT.get(classify_version(current_version), 5) for v in reversed(versions): if compare_versions(v, current_version) >= 0 and PRERELEASE_WEIGHT.get(classify_version(v), 5) >= current_rank: return v return None def get_upgrade_type(current: str, latest: str) -> str: if compare_versions(latest, current) <= 0: return "none" cur = _parse_segments(current) lat = _parse_segments(latest) max_len = max(len(cur), len(lat)) while len(cur) < max_len: cur.append(0) while len(lat) < max_len: lat.append(0) if lat[0] != cur[0]: return "major" if len(lat) > 1 and len(cur) > 1 and lat[1] != cur[1]: return "minor" if len(lat) > 2 and len(cur) > 2 and lat[2] != cur[2]: return "patch" return "patch" # --------------------------------------------------------------------------- # String distance & Solr query escaping # --------------------------------------------------------------------------- def _levenshtein(a: str, b: str) -> int: """Plain Levenshtein edit distance (insert/delete/substitute, each cost 1). NOT Damerau: an adjacent transposition is two substitutions, so ``_levenshtein("ab", "ba") == 2``. Iterative two-row DP, stdlib only. """ if a == b: return 0 if not a: return len(b) if not b: return len(a) prev = list(range(len(b) + 1)) for i, ca in enumerate(a, 1): cur = [i] for j, cb in enumerate(b, 1): cost = 0 if ca == cb else 1 cur.append(min(prev[j] + 1, cur[j - 1] + 1, prev[j - 1] + cost)) prev = cur return prev[-1] def _similarity(a: str, b: str) -> float: """Normalized edit-distance similarity in [0, 1]. ``1 - _levenshtein(a, b) / max(len(a), len(b), 1)``. Two empty strings score 1.0 (the ``max(..., 1)`` keeps the divisor non-zero so 0/1 = 0). Levenshtein never exceeds the longer length, so the result needs no clamping. """ return 1.0 - _levenshtein(a, b) / max(len(a), len(b), 1) # Lucene/Solr query metacharacters; ``&&`` / ``||`` operators are handled by # escaping each ``&`` / ``|``. Whitespace is escaped separately (below) so a # bareword operator like ``OR`` cannot survive as a whitespace-delimited token. _SOLR_SPECIAL = set('+-&|!(){}[]^"~*?:\\/') def _solr_escape(token: str) -> str: """Backslash-escape Solr/Lucene query metacharacters and whitespace. A token passed straight into a Solr query could otherwise be parsed as query syntax (``*`` wildcard, ``:`` field separator, a bareword ``OR``). Each special char and each whitespace char is prefixed with a backslash, so the whole token is interpreted literally as a single term. """ out = [] for ch in token: if ch in _SOLR_SPECIAL or ch.isspace(): out.append("\\") out.append(ch) return "".join(out) # --------------------------------------------------------------------------- # GitHub utilities # --------------------------------------------------------------------------- GITHUB_REPO_RE = re.compile(r"github\.com[/:]([A-Za-z0-9_.-]+)/([A-Za-z0-9_.-]+)") def _strip_xml_comments(xml: str) -> str: prev = None while prev != xml: prev = xml xml = re.sub(r"", "", xml) return xml def extract_github_repo_from_pom(pom_xml: str) -> Optional[Dict[str, str]]: xml = _strip_xml_comments(pom_xml) def _parse_github_url(url: str) -> Optional[Dict[str, str]]: m = GITHUB_REPO_RE.search(url) if not m: return None owner = m.group(1) repo = m.group(2).rstrip("/") repo = re.sub(r"\.git$", "", repo) repo = repo.split("/")[0] return {"owner": owner, "repo": repo} scm_match = re.search(r"([\s\S]*?)", xml) if scm_match: block = scm_match.group(1) for tag in ("url", "connection", "developerConnection"): m = re.search(rf"<{tag}>\s*(.*?)\s*", block) if m: r = _parse_github_url(m.group(1)) if r: return r # Fallback: root without_scm = re.sub(r"[\s\S]*?", "", xml) m = re.search(r"\s*(.*?)\s*", without_scm) if m: return _parse_github_url(m.group(1)) return None def extract_scm_url_from_pom(pom_xml: str) -> Optional[str]: xml = _strip_xml_comments(pom_xml) def _clean(v: str) -> str: return re.sub(r"^scm:[a-z]+:", "", v, flags=re.IGNORECASE) scm_match = re.search(r"([\s\S]*?)", xml) if scm_match: block = scm_match.group(1) for tag in ("url", "connection", "developerConnection"): m = re.search(rf"<{tag}>\s*(.*?)\s*", block) if m and m.group(1): return _clean(m.group(1).strip()) without_scm = re.sub(r"[\s\S]*?", "", xml) m = re.search(r"\s*(.*?)\s*", without_scm) if m and m.group(1): return m.group(1).strip() return None def extract_licenses_from_pom(pom_xml: str) -> List[Dict[str, Optional[str]]]: """Extract ```` entries from a POM (#300). Returns a list of ``{name, url}`` dicts. ``name`` comes from ````; ``url`` from ```` (either may be None when the tag is absent/empty). Entries with neither name nor url are skipped. Regex-only — no XML parser. """ xml = _strip_xml_comments(pom_xml) block_m = re.search(r"([\s\S]*?)", xml) if not block_m: return [] entries: List[Dict[str, Optional[str]]] = [] for m in re.finditer(r"([\s\S]*?)", block_m.group(1)): block = m.group(1) name_m = re.search(r"\s*(.*?)\s*", block) url_m = re.search(r"\s*(.*?)\s*", block) name = name_m.group(1).strip() if name_m and name_m.group(1) else None url = url_m.group(1).strip() if url_m and url_m.group(1) else None if name or url: entries.append({"name": name, "url": url}) return entries # --------------------------------------------------------------------------- # License intelligence (#300) — SPDX normalize, categorize, interpret # --------------------------------------------------------------------------- # Canonical SPDX id -> category. Static lookup only; no external license API. _LICENSE_CATEGORY_BY_SPDX: Dict[str, str] = { # permissive "MIT": "permissive", "Apache-2.0": "permissive", "Apache-1.1": "permissive", "Apache-1.0": "permissive", "BSD-2-Clause": "permissive", "BSD-3-Clause": "permissive", "BSD-3-Clause-Clear": "permissive", "BSD-4-Clause": "permissive", "ISC": "permissive", "0BSD": "permissive", "Unlicense": "permissive", "WTFPL": "permissive", "CC0-1.0": "permissive", "PostgreSQL": "permissive", "Zlib": "permissive", "BSL-1.0": "permissive", "NCSA": "permissive", "X11": "permissive", "Python-2.0": "permissive", "PSF-2.0": "permissive", "AFL-2.1": "permissive", "AFL-3.0": "permissive", "Artistic-2.0": "permissive", "MS-PL": "permissive", "PHP-3.01": "permissive", # weak copyleft "LGPL-2.0": "weak-copyleft", "LGPL-2.0-only": "weak-copyleft", "LGPL-2.0-or-later": "weak-copyleft", "LGPL-2.1": "weak-copyleft", "LGPL-2.1-only": "weak-copyleft", "LGPL-2.1-or-later": "weak-copyleft", "LGPL-3.0": "weak-copyleft", "LGPL-3.0-only": "weak-copyleft", "LGPL-3.0-or-later": "weak-copyleft", "MPL-1.1": "weak-copyleft", "MPL-2.0": "weak-copyleft", "EPL-1.0": "weak-copyleft", "EPL-2.0": "weak-copyleft", "CDDL-1.0": "weak-copyleft", "CDDL-1.1": "weak-copyleft", "CPL-1.0": "weak-copyleft", "IPL-1.0": "weak-copyleft", # strong copyleft "GPL-2.0": "strong-copyleft", "GPL-2.0-only": "strong-copyleft", "GPL-2.0-or-later": "strong-copyleft", "GPL-3.0": "strong-copyleft", "GPL-3.0-only": "strong-copyleft", "GPL-3.0-or-later": "strong-copyleft", # network copyleft "AGPL-3.0": "network-copyleft", "AGPL-3.0-only": "network-copyleft", "AGPL-3.0-or-later": "network-copyleft", "EUPL-1.1": "network-copyleft", "EUPL-1.2": "network-copyleft", } _LICENSE_CATEGORY_NOTES: Dict[str, str] = { "permissive": ( "Permissive open-source license: generally allows use, modification, " "and redistribution with attribution; no copyleft obligation on your code." ), "weak-copyleft": ( "Weak copyleft: modifications to this library itself may need to stay " "under the same license; linking from proprietary code is usually allowed." ), "strong-copyleft": ( "Strong copyleft: distributing a combined work that includes this " "library typically requires releasing that work under a compatible copyleft license." ), "network-copyleft": ( "Network copyleft: providing the software as a network service can " "trigger source-disclosure obligations similar to strong copyleft." ), "proprietary": ( "Declared license is not a recognized open-source SPDX id; treat as " "proprietary/custom and review the license text before adopting." ), "unknown": ( "No license was declared in the POM or on the linked GitHub repository; " "do not assume redistributable rights." ), } # Lowercased POM/GitHub license name fragments -> canonical SPDX. # Longer / more specific keys are matched first via sorted length. _LICENSE_NAME_TO_SPDX: Dict[str, str] = { "apache license, version 2.0": "Apache-2.0", "apache license version 2.0": "Apache-2.0", "the apache software license, version 2.0": "Apache-2.0", "apache software license, version 2.0": "Apache-2.0", "apache software license - version 2.0": "Apache-2.0", "asl 2.0": "Apache-2.0", "apache-2.0": "Apache-2.0", "apache 2.0": "Apache-2.0", "apache 2": "Apache-2.0", "apache license 2.0": "Apache-2.0", "the apache license, version 2.0": "Apache-2.0", "apache license, version 1.1": "Apache-1.1", "mit license": "MIT", "the mit license": "MIT", "mit": "MIT", "bsd 2-clause license": "BSD-2-Clause", "bsd 2-clause \"simplified\" license": "BSD-2-Clause", "bsd-2-clause": "BSD-2-Clause", "simplified bsd license": "BSD-2-Clause", "bsd 3-clause license": "BSD-3-Clause", "bsd 3-clause \"new\" or \"revised\" license": "BSD-3-Clause", "the bsd 3-clause license": "BSD-3-Clause", "bsd-3-clause": "BSD-3-Clause", "new bsd license": "BSD-3-Clause", "revised bsd license": "BSD-3-Clause", "isc license": "ISC", "isc": "ISC", "unlicense": "Unlicense", "wtfpl": "WTFPL", "cc0 1.0 universal": "CC0-1.0", "cc0-1.0": "CC0-1.0", "mozilla public license version 2.0": "MPL-2.0", "mozilla public license 2.0": "MPL-2.0", "mpl 2.0": "MPL-2.0", "mpl-2.0": "MPL-2.0", "eclipse public license - v 2.0": "EPL-2.0", "eclipse public license - v 1.0": "EPL-1.0", "eclipse public license 2.0": "EPL-2.0", "eclipse public license 1.0": "EPL-1.0", "epl-2.0": "EPL-2.0", "epl-1.0": "EPL-1.0", "common development and distribution license (cddl) v1.0": "CDDL-1.0", "common development and distribution license 1.0": "CDDL-1.0", "cddl-1.0": "CDDL-1.0", "cddl 1.1": "CDDL-1.1", "cddl-1.1": "CDDL-1.1", "gnu lesser general public license, version 3": "LGPL-3.0-only", "gnu lesser general public license version 3": "LGPL-3.0-only", "gnu lesser general public license, version 2.1": "LGPL-2.1-only", "gnu lesser general public license version 2.1": "LGPL-2.1-only", "lgpl-3.0": "LGPL-3.0-only", "lgpl 3.0": "LGPL-3.0-only", "lgplv3": "LGPL-3.0-only", "lgpl-2.1": "LGPL-2.1-only", "lgpl 2.1": "LGPL-2.1-only", "lgplv2.1": "LGPL-2.1-only", "gnu general public license, version 3": "GPL-3.0-only", "gnu general public license version 3": "GPL-3.0-only", "gnu general public license, version 2": "GPL-2.0-only", "gnu general public license version 2": "GPL-2.0-only", "gpl-3.0": "GPL-3.0-only", "gpl 3.0": "GPL-3.0-only", "gplv3": "GPL-3.0-only", "gpl-2.0": "GPL-2.0-only", "gpl 2.0": "GPL-2.0-only", "gplv2": "GPL-2.0-only", "gnu affero general public license v3": "AGPL-3.0-only", "gnu affero general public license version 3": "AGPL-3.0-only", "agpl-3.0": "AGPL-3.0-only", "agpl 3.0": "AGPL-3.0-only", "agplv3": "AGPL-3.0-only", "european union public licence 1.2": "EUPL-1.2", "european union public license 1.2": "EUPL-1.2", "eupl-1.2": "EUPL-1.2", "eupl 1.2": "EUPL-1.2", "zlib license": "Zlib", "boost software license 1.0": "BSL-1.0", "bsl-1.0": "BSL-1.0", } _LICENSE_NAME_TO_SPDX_KEYS = sorted(_LICENSE_NAME_TO_SPDX.keys(), key=len, reverse=True) def normalize_license_to_spdx(raw: Optional[str]) -> Optional[str]: """Map a POM/GitHub license string to a canonical SPDX id, or None.""" if not raw: return None text = raw.strip() if not text or text.upper() in ("NOASSERTION", "NONE", "UNKNOWN", "SEE LICENSE", "SEE LICENSE IN LICENSE"): return None if text in _LICENSE_CATEGORY_BY_SPDX: return text # GitHub sometimes returns SPDX with "NOASSERTION"; already handled above. lower = re.sub(r"\s+", " ", text.lower()) lower = lower.strip(" .;") if lower in _LICENSE_NAME_TO_SPDX: return _LICENSE_NAME_TO_SPDX[lower] for key in _LICENSE_NAME_TO_SPDX_KEYS: if key in lower: return _LICENSE_NAME_TO_SPDX[key] # Bare SPDX-looking token (e.g. already-canonical from GitHub). token = text.strip() if token in _LICENSE_CATEGORY_BY_SPDX: return token return None def categorize_license(spdx_id: Optional[str], raw_name: Optional[str] = None) -> str: """Return license category for an SPDX id / raw name (#300).""" if spdx_id and spdx_id in _LICENSE_CATEGORY_BY_SPDX: return _LICENSE_CATEGORY_BY_SPDX[spdx_id] if spdx_id or (raw_name and raw_name.strip()): return "proprietary" return "unknown" def license_category_notes(category: str) -> str: return _LICENSE_CATEGORY_NOTES.get(category, _LICENSE_CATEGORY_NOTES["unknown"]) def _license_result( group_id: str, artifact_id: str, version: Optional[str], spdx_id: Optional[str], name: Optional[str], url: Optional[str], category: str, source: Optional[str], error: Optional[str] = None, ) -> Dict[str, Any]: out: Dict[str, Any] = { "groupId": group_id, "artifactId": artifact_id, "spdxId": spdx_id, "name": name, "url": url, "category": category, "notes": license_category_notes(category), "source": source, } if version is not None: out["version"] = version if error is not None: out["error"] = error return out def resolve_dependency_license( group_id: str, artifact_id: str, version: Optional[str], ctx: "ResolutionContext", ) -> Dict[str, Any]: """Resolve license intelligence for one GAV (#300). Reuses the health-tool path: metadata → POM licenses → optional GitHub ``license.spdx_id``. Category/notes come from static tables only. """ resolved_version = version resolved_from = None try: metadata = fetch_metadata(group_id, artifact_id, ctx) resolved_from = metadata.get("resolvedFrom") if not resolved_version: versions = metadata.get("versions") or [] resolved_version = find_latest_version(versions, "PREFER_STABLE") or ( versions[-1] if versions else None ) if not resolved_version: result = _license_result( group_id, artifact_id, None, None, None, None, "unknown", None, error="No version found", ) if resolved_from is not None: result["resolvedFrom"] = resolved_from return result except Exception as e: return _license_result( group_id, artifact_id, version, None, None, None, "unknown", None, error=str(e), ) pom_name: Optional[str] = None pom_url: Optional[str] = None pom_xml = fetch_pom(group_id, artifact_id, resolved_version, ctx) if resolved_version else None if pom_xml: pom_entries = extract_licenses_from_pom(pom_xml) if pom_entries: pom_name = pom_entries[0]["name"] pom_url = pom_entries[0]["url"] github_spdx: Optional[str] = None gh_repo = extract_github_repo_from_pom(pom_xml) if pom_xml else None if not gh_repo: guess = guess_github_repo(group_id, artifact_id) if guess: # Prefer a cheap existence check before fetching full repo metadata, # matching get_dependency_health's guess path cost profile. cached = gh_fetch_repo(guess["owner"], guess["repo"]) if cached: gh_repo = guess spdx = (cached.get("license") or {}).get("spdx_id") if spdx and spdx != "NOASSERTION": github_spdx = spdx if gh_repo and github_spdx is None: repo_meta = gh_fetch_repo(gh_repo["owner"], gh_repo["repo"]) if repo_meta: spdx = (repo_meta.get("license") or {}).get("spdx_id") if spdx and spdx != "NOASSERTION": github_spdx = spdx # Prefer GitHub SPDX (already canonical) over POM name normalization. source: Optional[str] = None spdx_id: Optional[str] = None name: Optional[str] = pom_name url: Optional[str] = pom_url if github_spdx: spdx_id = normalize_license_to_spdx(github_spdx) or github_spdx source = "github" if not name: name = github_spdx elif pom_name: normalized = normalize_license_to_spdx(pom_name) if normalized: spdx_id = normalized source = "pom" if pom_name in _LICENSE_CATEGORY_BY_SPDX else "spdx-normalized" else: source = "pom" else: source = None category = categorize_license(spdx_id, name or github_spdx) result = _license_result( group_id, artifact_id, resolved_version, spdx_id, name, url, category, source, ) if resolved_from is not None: result["resolvedFrom"] = resolved_from return result # --------------------------------------------------------------------------- # Transitive license compliance (#289) # --------------------------------------------------------------------------- # Categories treated as risky under a permissive projectLicense posture. # Overridable via ``disallow`` (SPDX ids and/or category names). _DEFAULT_DISALLOW_FOR_PERMISSIVE = frozenset({ "strong-copyleft", "network-copyleft", "proprietary", }) # Known category tokens accepted in ``disallow`` (case-insensitive match). _LICENSE_CATEGORY_TOKENS = frozenset({ "permissive", "weak-copyleft", "strong-copyleft", "network-copyleft", "proprietary", "unknown", }) _LICENSE_COMPLIANCE_NOTES = [ ( "License data comes from deps.dev GetVersion (package metadata SPDX " "expressions), not from a full legal review of license text." ), ( "GetDependencies does not include licenses; each unique GAV in the " "resolved graph requires a separate GetVersion call (cached, capped)." ), ( "Graphs are resolved per root in isolation via deps.dev — not a full " "Maven/Gradle project resolve (exclusions, dependencyManagement, " "ResolutionStrategy, private coordinates are not modeled)." ), ( "Verdicts are heuristic policy signals, not legal advice. Independently " "verify licenses before redistributing." ), ( "deps.dev may return SPDX expressions (e.g. Apache-2.0 OR MIT) or " "'non-standard'; expression operators beyond a single id are treated " "conservatively (review when not an exact known SPDX id)." ), ] def _normalize_disallow_token(raw: str) -> str: """Normalize a disallow entry to a category name or canonical SPDX id.""" text = (raw or "").strip() if not text: return "" lower = text.lower() if lower in _LICENSE_CATEGORY_TOKENS: return lower # Accept hyphen/underscore variants of category names. compact = lower.replace("_", "-") if compact in _LICENSE_CATEGORY_TOKENS: return compact spdx = normalize_license_to_spdx(text) return spdx or text def resolve_license_policy( project_license: Optional[str] = None, disallow: Optional[List[str]] = None, ) -> Dict[str, Any]: """Build the effective disallow set for license compliance (#289). Default posture when ``projectLicense`` is permissive (or omitted with no custom ``disallow``): flag strong-copyleft, network-copyleft, and proprietary. Explicit ``disallow`` replaces the default set entirely. Unknown project licenses do not invent a default disallow list unless ``disallow`` is provided. """ project_spdx = normalize_license_to_spdx(project_license) if project_license else None project_category = ( categorize_license(project_spdx, project_license) if (project_spdx or project_license) else None ) if disallow is not None: tokens = [] seen = set() for item in disallow: tok = _normalize_disallow_token(str(item)) if tok and tok not in seen: seen.add(tok) tokens.append(tok) source = "custom" elif project_category == "permissive" or ( project_license is None and project_category is None ): # Omitted projectLicense → assume permissive posture (common OSS default # for this tool's audience). Explicit non-permissive projectLicense # without disallow → empty set (no invented policy). tokens = sorted(_DEFAULT_DISALLOW_FOR_PERMISSIVE) source = "default-permissive" else: tokens = [] source = "none" return { "projectLicense": project_spdx or project_license, "projectCategory": project_category, "disallow": tokens, "policySource": source, } def _is_spdx_expression(raw: str) -> bool: """True when ``raw`` looks like a compound SPDX expression, not a single id. deps.dev may return ``Apache-2.0 OR MIT``. Substring normalization would otherwise pick the first known id and falsely categorize as permissive. """ upper = f" {raw.upper()} " return ( " OR " in upper or " AND " in upper or " WITH " in upper or raw.strip().startswith("(") ) def _primary_license_from_depsdev(raw_licenses: List[str]) -> Dict[str, Any]: """Pick a primary SPDX/category from deps.dev license strings. Multi-license lists and SPDX expressions that are not a single known id degrade to category ``unknown`` with the raw expression preserved, so the compliance verdict becomes ``review`` rather than a false ``ok``. """ if not raw_licenses: return { "spdxId": None, "name": None, "category": "unknown", "licenses": [], } # Prefer the first entry that is a single known SPDX id. for raw in raw_licenses: if raw.lower() == "non-standard": return { "spdxId": None, "name": raw, "category": "proprietary", "licenses": list(raw_licenses), } if _is_spdx_expression(raw): continue spdx = normalize_license_to_spdx(raw) if spdx and spdx in _LICENSE_CATEGORY_BY_SPDX and spdx == raw.strip(): return { "spdxId": spdx, "name": spdx, "category": categorize_license(spdx, raw), "licenses": list(raw_licenses), } if spdx and spdx in _LICENSE_CATEGORY_BY_SPDX and not _is_spdx_expression(raw): # Exact table hit or simple alias (e.g. "MIT License") — not an expression. return { "spdxId": spdx, "name": raw if raw != spdx else spdx, "category": categorize_license(spdx, raw), "licenses": list(raw_licenses), } # Expression / unrecognized — keep raw, force review via unknown. primary = raw_licenses[0] return { "spdxId": None, "name": primary, "category": "unknown", "licenses": list(raw_licenses), } def license_compliance_verdict( *, spdx_id: Optional[str], category: str, policy: Dict[str, Any], fetch_error: Optional[str] = None, missing_license: bool = False, ) -> Dict[str, str]: """Return ``{verdict, reason}`` for one node against the policy. - ``violation`` — category or SPDX id is in the disallow set. - ``review`` — missing/unknown license metadata, or fetch failure. - ``ok`` — known license not disallowed. """ disallow = set(policy.get("disallow") or []) if fetch_error: return { "verdict": "review", "reason": f"license metadata unavailable: {fetch_error}", } if missing_license or category == "unknown" or not spdx_id: # Unknown / empty must never be a false ok (AC). if category in disallow or "unknown" in disallow: return { "verdict": "violation", "reason": f"disallowed category '{category}' (license unknown or undeclared)", } return { "verdict": "review", "reason": "no SPDX license declared by deps.dev; verify manually", } if category in disallow: return { "verdict": "violation", "reason": f"category '{category}' is disallowed by policy", } if spdx_id in disallow: return { "verdict": "violation", "reason": f"SPDX id '{spdx_id}' is disallowed by policy", } return {"verdict": "ok", "reason": "license allowed by policy"} def check_license_compliance( dependencies: List[Dict[str, Any]], project_license: Optional[str] = None, disallow: Optional[List[str]] = None, ) -> Dict[str, Any]: """Aggregate transitive licenses and flag policy violations (#289). For each versioned root, fetches the deps.dev GetDependencies graph, then GetVersion licenses for each unique GAV (capped). Marks ``viaTransitive`` from deps.dev ``relation`` (SELF/DIRECT → false, else true). #400/#402 concurrency note: the per-ROOT graph fetch below (bounded by MAX_LICENSE_COMPLIANCE_ROOTS, a pure independent network call per root) IS parallelized. The per-NODE license fetch/dedup walk (_ensure_license / _add_result, up to MAX_LICENSE_COMPLIANCE_NODES) deliberately stays SEQUENTIAL: its dedup is order-sensitive (`seen_result_keys`'s "first-seen wins for viaTransitive=false, upgrade in place otherwise" rule depends on processing roots/nodes in a fixed order) and its cache (`license_cache`) is mutated in lockstep with that walk — splitting fetch from decide to parallelize it safely would be a materially larger, higher-risk rewrite of this function's carefully-specified dedup semantics. Instead, TOOL_DEADLINE bounds the sequential node walk directly (see the `_now() >= deadline` checks below): a degenerate large-graph batch returns whatever was gathered so far with `partial: true` rather than blocking indefinitely — which is exactly what #402 asks for on this specific tool. """ deadline = _now() + TOOL_DEADLINE deadline_hit = False policy = resolve_license_policy(project_license, disallow) roots = [] for dep in dependencies or []: gid = dep.get("groupId") aid = dep.get("artifactId") ver = dep.get("version") if not gid or not aid or not ver: continue roots.append({"groupId": gid, "artifactId": aid, "version": ver}) truncated_roots = len(roots) > MAX_LICENSE_COMPLIANCE_ROOTS roots = roots[:MAX_LICENSE_COMPLIANCE_ROOTS] # Dedup GetVersion calls across roots: g:a:v → license fetch result. license_cache: Dict[str, Dict[str, Any]] = {} results: List[Dict[str, Any]] = [] seen_result_keys: set = set() errors: List[str] = [] partial = False truncated_nodes = False nodes_fetched = 0 capability: Optional[str] = None def _gav_key(g: str, a: str, v: str) -> str: return f"{g}:{a}:{v}" def _ensure_license(g: str, a: str, v: str) -> Dict[str, Any]: nonlocal nodes_fetched, truncated_nodes, partial, capability key = _gav_key(g, a, v) if key in license_cache: return license_cache[key] if nodes_fetched >= MAX_LICENSE_COMPLIANCE_NODES: truncated_nodes = True partial = True entry: Dict[str, Any] = { "ok": False, "licenses": [], "error": ( f"GetVersion fan-out capped at {MAX_LICENSE_COMPLIANCE_NODES}" ), } license_cache[key] = entry return entry nodes_fetched += 1 entry = fetch_depsdev_licenses(g, a, v) if entry.get("capabilityUnavailable") and capability is None: capability = entry["capabilityUnavailable"] license_cache[key] = entry return entry def _add_result( g: str, a: str, v: str, *, via_transitive: bool, relation: str, root: Dict[str, str], ) -> None: nonlocal partial key = _gav_key(g, a, v) # One row per unique GAV across all roots (first-seen wins for # viaTransitive=false — a direct hit beats a later transitive sighting). if key in seen_result_keys: # Upgrade viaTransitive False if we later see it as direct/self. if not via_transitive: for row in results: if ( row["groupId"] == g and row["artifactId"] == a and row["version"] == v and row.get("viaTransitive") ): row["viaTransitive"] = False row["relation"] = relation or row.get("relation") break return seen_result_keys.add(key) fetched = _ensure_license(g, a, v) primary = _primary_license_from_depsdev(fetched.get("licenses") or []) missing = not (fetched.get("licenses") or []) verdict_info = license_compliance_verdict( spdx_id=primary.get("spdxId"), category=primary.get("category") or "unknown", policy=policy, fetch_error=None if fetched.get("ok") else fetched.get("error"), missing_license=missing and fetched.get("ok", False), ) if verdict_info["verdict"] != "ok": # review/violation are expected outcomes; do not mark partial solely # for policy hits. Partial is reserved for data/transport gaps. pass if not fetched.get("ok"): partial = True row: Dict[str, Any] = { "groupId": g, "artifactId": a, "version": v, "spdxId": primary.get("spdxId"), "license": primary.get("spdxId") or primary.get("name"), "licenses": primary.get("licenses") or [], "category": primary.get("category") or "unknown", "viaTransitive": via_transitive, "relation": relation or ("INDIRECT" if via_transitive else "DIRECT"), "verdict": verdict_info["verdict"], "reason": verdict_info["reason"], "root": { "groupId": root["groupId"], "artifactId": root["artifactId"], "version": root["version"], }, "source": "deps.dev", } if fetched.get("error"): row["error"] = fetched["error"] results.append(row) # #400: fetch every root's transitive graph in parallel first (pure, # independent per-root network calls; no shared state touched) — then # walk roots+nodes SEQUENTIALLY below in the ORIGINAL order, exactly as # before, so the dedup/upgrade-in-place logic in _add_result stays # deterministic regardless of which root's fetch actually completed # first. def _fetch_root_graph(root: Dict[str, str]) -> Dict[str, Any]: return fetch_depsdev_dependencies( root["groupId"], root["artifactId"], root["version"], ) root_graphs, root_fetch_partial = _map_parallel( roots, _fetch_root_graph, max_workers=MAX_PARALLEL_FETCHES, deadline=deadline, ) if root_fetch_partial: deadline_hit = True for root, fetched in zip(roots, root_graphs): # R2a fix: this loop must NEVER `break` out early on a deadline — # doing so silently drops every root from that point on (the exact # bug: the deadline check used to fire immediately after synthesizing # a cut-off root's placeholder, before that root's own review row was # ever added, then `break` skipped ALL subsequent roots too, since a # once-exceeded deadline never un-expires). Instead: once the # deadline is exceeded — whether THIS root's own graph fetch was cut # off (`fetched is None`) or time simply ran out by the time we # reached this iteration (a prior root's node walk used the rest of # the budget) — this root degrades to a cheap review row (no further # network calls) and the loop continues to the NEXT root exactly the # same way. Every root in `roots` therefore always reaches the # bottom of one iteration and gets exactly one row. deadline_cutoff = fetched is None or _now() >= deadline if deadline_cutoff: deadline_hit = True cap = fetched.get("capabilityUnavailable") if fetched is not None else None if cap and capability is None: capability = cap if deadline_cutoff: err = (fetched or {}).get("error") if fetched is not None else None fetched = { "ok": False, "error": err or f"skipped: exceeded the {TOOL_DEADLINE}s tool deadline", } if not fetched.get("ok"): partial = True err = fetched.get("error") or "deps.dev unavailable" if len(errors) < MAX_DEPSDEV_ERRORS_REPORTED: errors.append( f"{root['groupId']}:{root['artifactId']}:{root['version']}: {err}" ) # Still emit a review row for the root itself so callers see it. _add_result( root["groupId"], root["artifactId"], root["version"], via_transitive=False, relation="SELF", root=root, ) # Force review on the synthetic root row when graph fetch failed. for row in results: if ( row["groupId"] == root["groupId"] and row["artifactId"] == root["artifactId"] and row["version"] == root["version"] ): if row["verdict"] == "ok": row["verdict"] = "review" row["reason"] = f"transitive graph unavailable: {err}" break continue if fetched.get("graphError"): partial = True if len(errors) < MAX_DEPSDEV_ERRORS_REPORTED: errors.append( f"{root['groupId']}:{root['artifactId']}:{root['version']}: " f"graph error: {fetched['graphError']}" ) if fetched.get("truncated") or fetched.get("partial"): partial = True for node in fetched.get("nodes") or []: if _now() >= deadline: # #402: bound the sequential per-node dedup/license-fetch # walk — see the concurrency note in this function's # docstring for why this loop stays sequential rather than # joining the #400 executor. Breaks only the NODE loop for # THIS root (nodes already added stay); the OUTER root loop # continues naturally to the next root, which will see # `deadline_cutoff=True` at the top of its own iteration and # degrade to a review row instead of attempting its nodes. deadline_hit = True break g = node.get("groupId") or "" a = node.get("artifactId") or "" v = node.get("version") or "" if not g or not a or not v: continue relation = (node.get("relation") or "").upper() via = relation not in ("SELF", "DIRECT", "") # Empty relation on the first node is typically SELF; treat unknown # non-empty as transitive to avoid under-flagging. if relation == "": via = False _add_result(g, a, v, via_transitive=via, relation=relation or "DIRECT", root=root) by_verdict = {"ok": 0, "review": 0, "violation": 0} by_category: Dict[str, int] = {} for row in results: v = row.get("verdict") or "review" by_verdict[v] = by_verdict.get(v, 0) + 1 cat = row.get("category") or "unknown" by_category[cat] = by_category.get(cat, 0) + 1 notes = list(_LICENSE_COMPLIANCE_NOTES) if truncated_roots: partial = True notes.append( f"Roots truncated to {MAX_LICENSE_COMPLIANCE_ROOTS} " f"(MAX_LICENSE_COMPLIANCE_ROOTS); results are partial." ) if truncated_nodes: notes.append( f"GetVersion calls capped at {MAX_LICENSE_COMPLIANCE_NODES} " f"(MAX_LICENSE_COMPLIANCE_NODES); results are partial." ) if deadline_hit: partial = True notes.append( f"Stopped early after exceeding the {TOOL_DEADLINE}s tool " f"deadline; remaining roots/nodes were not processed and " f"results are partial." ) out: Dict[str, Any] = { "policy": policy, "summary": { "total": len(results), "byVerdict": by_verdict, "byCategory": by_category, "violationCount": by_verdict.get("violation", 0), "reviewCount": by_verdict.get("review", 0), }, "results": results, "partial": partial, "notes": notes, } if errors: out["errors"] = errors return _with_capability(out, capability) def extract_relocation_from_pom( pom_xml: str, group_id: str, artifact_id: str, version: str ) -> Optional[Dict[str, str]]: """Extract Maven artifact relocation (``) from a POM (#284). Returns None when there is no `` block. Per Maven's relocation spec, any of groupId/artifactId/version absent inside `` means "unchanged from the original coordinate" for that field — e.g. a relocation POM specifying only a new `` means the groupId/version carry over. The missing fields are filled in from the original coordinate here so callers get a complete, directly-usable coordinate rather than a partial one they would have to merge themselves. The `` search is scoped to inside `` only — Maven Shade Plugin's `` (package relocation for shading, an unrelated concept) also uses the `` tag, so an unscoped search would false-positive on a POM that shades dependencies but never relocates its own coordinates.""" xml = _strip_xml_comments(pom_xml) dm_m = re.search( r"([\s\S]*?)", xml ) if not dm_m: return None reloc_m = re.search(r"([\s\S]*?)", dm_m.group(1)) if not reloc_m: return None block = reloc_m.group(1) gid_m = _extract_tag(block, "groupId") aid_m = _extract_tag(block, "artifactId") ver_m = _extract_tag(block, "version") return { "groupId": gid_m.strip() if gid_m else group_id, "artifactId": aid_m.strip() if aid_m else artifact_id, "version": ver_m.strip() if ver_m else version, } def _gradle_plugin_marker_plugin_id(group_id: str, artifact_id: str) -> Optional[str]: """Returns the plugin id if (group_id, artifact_id) is a Gradle plugin marker coordinate (`{pluginId}:{pluginId}.gradle.plugin`), else None. Stricter than the `.gradle.plugin`-suffix-only check used for repo-scope routing (_repos_for) — that check stays suffix-only by design (see AGENTS.md); this one additionally requires group_id == pluginId, which is the actual marker shape, since it drives POM-dependency resolution, not just repo routing.""" suffix = ".gradle.plugin" if not artifact_id.endswith(suffix): return None plugin_id = artifact_id[: -len(suffix)] if not plugin_id or group_id != plugin_id: return None return plugin_id def resolve_plugin_marker_implementation( group_id: str, artifact_id: str, version: Optional[str], ctx: "ResolutionContext" ) -> Optional[Dict[str, str]]: """Gradle plugin marker artifacts are minimal POMs whose only purpose is a single pointing at the real implementation artifact — OSV indexes the implementation, not the marker, so callers must resolve markers before querying vulnerabilities (#290). Returns {groupId, artifactId, version} of the resolved implementation, or None when the coordinate isn't a marker, or resolution fails for any reason (missing version, POM fetch failure, no/ incomplete block, unresolved ${...} property) — callers MUST degrade gracefully on None, never raise.""" if not _gradle_plugin_marker_plugin_id(group_id, artifact_id): return None if not version: return None pom = fetch_pom(group_id, artifact_id, version, ctx) if not pom: return None xml = _strip_xml_comments(pom) # Drop any block before searching: it holds version # pins, not the marker's real implementation dependency, and an unscoped # search would match a inside it first if one is present. # NOTE: duplicates _parse_maven_deps' dependency-block extraction (~1508) # with slightly different field requirements; not unified to keep this fix # minimal — candidate for a future shared single-dependency-block parser. xml = re.sub(r"[\s\S]*?", "", xml) dep_m = re.search(r"([\s\S]*?)", xml) if not dep_m: return None block = dep_m.group(1) gid_m = _extract_tag(block, "groupId") aid_m = _extract_tag(block, "artifactId") ver_m = _extract_tag(block, "version") if not gid_m or not aid_m or not ver_m: return None impl_group_id = gid_m.strip() impl_artifact_id = aid_m.strip() impl_version = ver_m.strip() if not impl_group_id or not impl_artifact_id or not impl_version: return None if impl_version.startswith("${"): return None return {"groupId": impl_group_id, "artifactId": impl_artifact_id, "version": impl_version} def guess_github_repo(group_id: str, artifact_id: str) -> Optional[Dict[str, str]]: for prefix in ("com.github.", "io.github."): if group_id.startswith(prefix) and len(group_id) > len(prefix): rest = group_id[len(prefix):] owner = rest.split(".")[0] return {"owner": owner, "repo": artifact_id} return None def scm_host(url: str) -> str: if not url: return "other" scp_m = re.match(r"^[^/@]+@([^:/]+):", url) if scp_m: host = scp_m.group(1).lower() else: m = re.match(r"https?://([^/]+)", url) host = m.group(1).lower() if m else "" if host in ("github.com",) or host.endswith(".github.com"): return "github" if host in ("gitlab.com",) or host.endswith(".gitlab.com"): return "gitlab" if host in ("bitbucket.org",) or host.endswith(".bitbucket.org"): return "bitbucket" return "other" def _gh_get(path: str) -> Optional[Any]: if _external_capability("github"): return None if not path.startswith("/"): path = "/" + path url = f"{_github_api_base()}{path}" try: status, body = http_get( url, _github_headers(), timeout=HTTP_TIMEOUT_EXTERNAL ) if status == 200: return json.loads(body) except Exception: pass # Network or parse error — caller treats None as "unavailable" return None def gh_repo_exists(owner: str, repo: str) -> bool: if _external_capability("github"): return False url = f"{_github_api_base()}/repos/{owner}/{repo}" try: status, _ = http_get( url, _github_headers(), timeout=HTTP_TIMEOUT_EXTERNAL ) return status == 200 except Exception: return False def gh_fetch_repo(owner: str, repo: str) -> Optional[Dict]: return _gh_get(f"/repos/{owner}/{repo}") def gh_fetch_releases(owner: str, repo: str) -> List[Dict]: data = _gh_get(f"/repos/{owner}/{repo}/releases?per_page=100") return data if isinstance(data, list) else [] def gh_fetch_user(login: str) -> Optional[Dict]: return _gh_get(f"/users/{login}") def gh_fetch_issue_stats(owner: str, repo: str) -> Optional[Dict]: def search_count(state: str) -> Optional[int]: q = urllib.parse.quote(f"repo:{owner}/{repo} type:issue state:{state}") data = _gh_get(f"/search/issues?q={q}&per_page=1") if data and isinstance(data.get("total_count"), int): return data["total_count"] return None def median_days_to_close() -> Optional[int]: q = urllib.parse.quote(f"repo:{owner}/{repo} type:issue state:closed") data = _gh_get(f"/search/issues?q={q}&sort=updated&order=desc&per_page=30") if not data or not isinstance(data.get("items"), list): return None durations = [] for item in data["items"]: ca = item.get("created_at") cl = item.get("closed_at") if ca and cl: try: t_created = _parse_iso(ca) t_closed = _parse_iso(cl) diff = (t_closed - t_created).total_seconds() if diff >= 0: durations.append(diff) except Exception: pass # Skip malformed date strings in issue timeline if not durations: return None durations.sort() mid = len(durations) // 2 if len(durations) % 2 == 1: median_sec = durations[mid] else: median_sec = (durations[mid - 1] + durations[mid]) / 2 return round(median_sec / 86400) open_count = search_count("open") closed_count = search_count("closed") if open_count is None and closed_count is None: return None total = (open_count or 0) + (closed_count or 0) close_ratio = (closed_count / total) if (open_count is not None and closed_count is not None and total > 0) else None median_days = median_days_to_close() return { "open": open_count, "closed": closed_count, "closeRatio": close_ratio, "medianDaysToClose": median_days, } def _parse_iso(s: str): import datetime # Handle Z suffix s = s.replace("Z", "+00:00") try: return datetime.datetime.fromisoformat(s) except Exception: # Fallback for Python 3.6 which doesn't support +00:00 in fromisoformat return datetime.datetime.strptime(s[:19], "%Y-%m-%dT%H:%M:%S") def _months_since(iso: Optional[str]) -> Optional[int]: if not iso: return None try: import datetime dt = _parse_iso(iso) # datetime.utcnow() is deprecated on Python 3.12+; now(timezone.utc) is the # replacement. tzinfo is stripped so `now` stays offset-naive: `dt` is forced # naive just below, and subtracting a naive from an aware datetime raises # TypeError. The wall-clock UTC value is identical to the old utcnow(), so the # months delta is unchanged. now = datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) # Make both offset-naive for comparison if hasattr(dt, 'utcoffset') and dt.utcoffset() is not None: dt = dt.replace(tzinfo=None) delta = now - dt return int(delta.days / 30) except Exception: return None def _summarize_releases(releases: List[Dict]) -> Dict: times = [] for r in releases: if r.get("draft") or r.get("prerelease"): continue pub = r.get("published_at") if not pub: continue try: dt = _parse_iso(pub) times.append((dt, pub)) except Exception: pass # Skip releases with unparseable published_at dates times.sort(key=lambda x: x[0], reverse=True) count = len(times) if count == 0: return {"last": None, "cadenceDays": None, "count": 0} last = times[0][1] if count < 2: return {"last": last, "cadenceDays": None, "count": count} gaps = [] for i in range(len(times) - 1): diff = (times[i][0] - times[i + 1][0]).total_seconds() gaps.append(diff) gaps.sort() mid = len(gaps) // 2 if len(gaps) % 2 == 1: median_sec = gaps[mid] else: median_sec = (gaps[mid - 1] + gaps[mid]) / 2 return {"last": last, "cadenceDays": round(median_sec / 86400), "count": count} def discover_github_repo(group_id: str, artifact_id: str, version: str, ctx: "ResolutionContext") -> Optional[Dict[str, str]]: """Try POM SCM, then guess from groupId. Returns {owner, repo} or None.""" pom = fetch_pom(group_id, artifact_id, version, ctx) if pom: r = extract_github_repo_from_pom(pom) if r: return r guess = guess_github_repo(group_id, artifact_id) if guess and gh_repo_exists(guess["owner"], guess["repo"]): return guess return None # --------------------------------------------------------------------------- # Changelog providers for get_dependency_changes (#308) # Order: AndroidX docs → AGP docs → GitHub releases (mirrors retired TS resolver). # --------------------------------------------------------------------------- AGP_GROUP_ID = "com.android.tools.build" AGP_RELEASES_BASE = "https://developer.android.com/build/releases" ANDROIDX_RELEASES_BASE = "https://developer.android.com/jetpack/androidx/releases" # Release-notes HTML pages change infrequently; match the retired TS 7-day TTL. TTL_CHANGELOG_HTML = TTL_POM def _agp_releases_base() -> str: return f"{_android_docs_base()}/build/releases" def _androidx_releases_base() -> str: return f"{_android_docs_base()}/jetpack/androidx/releases" _AGP_HEADING_RE = re.compile( r']*\s+data-text="Android Gradle plugin ([\d][^\s"]*)"[^>]*>', re.IGNORECASE, ) _ANDROIDX_VERSION_HEADING_RE = re.compile( r"]*>\s*Version\s+([\d][^\s<]*)\s*", re.IGNORECASE, ) def html_to_text(html: str) -> str: """Minimal HTML→text for Android docs release notes (stdlib only, #308). Mirrors the retired TS ``html/to-text``: list/br/p formatting, iterative tag strip (handles smuggled nested tags), then a small entity unescape set. """ formatted = re.sub(r"]*>", "- ", html, flags=re.IGNORECASE) formatted = re.sub(r"", "\n", formatted, flags=re.IGNORECASE) formatted = re.sub(r"", "\n", formatted, flags=re.IGNORECASE) formatted = re.sub(r"

", "\n\n", formatted, flags=re.IGNORECASE) # Loop so forms like ``<