ariadne/ariadne/services/hermes_sonar_client.py

"""Read open SonarQube findings for a project.

Triage has so far been incident-driven: something fails, and the failure is
what enters the flow. Static analysis is the opposite shape - a standing
backlog that never fails a build and therefore never asks anyone for
attention. On this instance that backlog is 139 open findings on Ariadne
alone, each one already naming its file, its line, its rule and what is wrong.
That is better-located evidence than the console text the code-repair flow
usually works from, and it is being thrown away.

This module only reads. It fetches findings and normalizes them into the same
shape the patch flow already consumes; deciding which are worth acting on
belongs to the sweep, and changing anything belongs to Ariadne.

Two things are deliberately excluded rather than filtered later:

Security hotspots are not findings. SonarQube models them as "needs a human to
look at this", and the resolution is a review decision, not a code change. The
quality gate on this instance fails on exactly that condition, and an
automation that "fixed" it would be marking hotspots reviewed without review -
defeating the control rather than satisfying it. They live behind a different
endpoint, and this module does not call it.

Findings already resolved or marked won't-fix carry a human's judgement. Asking
a model to reopen that question wastes tokens and produces pull requests that
argue with a decision someone already made.
"""

from __future__ import annotations

import base64
from typing import Any

import httpx

from ..utils.logging import get_logger


logger = get_logger(__name__)

HTTP_OK = 200

CODE_SMELL = "CODE_SMELL"
BUG = "BUG"
VULNERABILITY = "VULNERABILITY"

# Hotspots are absent on purpose; see the module docstring.
ALL_ISSUE_TYPES: tuple[str, ...] = (CODE_SMELL, BUG, VULNERABILITY)

_DEFAULT_TIMEOUT_SECONDS = 20.0
_DEFAULT_PAGE_SIZE = 100
_MAX_PAGE_SIZE = 500
# Only findings nobody has ruled on. `resolved=false` already excludes fixed
# ones; this excludes the ones a person decided to keep.
_OPEN_STATUSES = "OPEN,CONFIRMED,REOPENED"


def fetch_issues(cfg: dict, project: str) -> tuple[list[dict[str, Any]], str | None]:
    """Fetch one project's open findings from SonarQube.

    Inputs: `cfg` with sonar_base_url, sonar_token, optionally sonar_types,
    sonar_severities and timeout_seconds; the SonarQube project key. Outputs:
    (issues, error) with exactly one side meaningful - issues is [] on error.

    Never raises and never logs the token. The token is sent as HTTP basic
    with an empty password, which is how SonarQube accepts user tokens.
    """

    base_url = str(cfg.get("sonar_base_url") or "").rstrip("/")
    if not base_url:
        return [], "sonar base url is empty"
    if not str(cfg.get("sonar_token") or ""):
        return [], "sonar token is empty"
    params = {
        "componentKeys": project,
        "resolved": "false",
        "statuses": _OPEN_STATUSES,
        "types": ",".join(_types(cfg)),
        "ps": str(_page_size(cfg)),
        "p": "1",
    }
    severities = _severities(cfg)
    if severities:
        params["severities"] = ",".join(severities)
    try:
        with httpx.Client(timeout=_timeout(cfg)) as client:
            response = client.get(
                f"{base_url}/api/issues/search", headers=_headers(cfg), params=params
            )
    except Exception as exc:
        return [], f"sonar fetch failed: {exc}"
    if response.status_code != HTTP_OK:
        return [], f"sonar fetch http {response.status_code}"
    try:
        payload = response.json()
    except Exception as exc:
        return [], f"sonar response not json: {exc}"
    return _normalized(payload), None


def _normalized(payload: Any) -> list[dict[str, Any]]:
    """Reduce the API payload to the fields the patch flow can act on."""

    issues = payload.get("issues") if isinstance(payload, dict) else None
    found: list[dict[str, Any]] = []
    for issue in issues if isinstance(issues, list) else []:
        if not isinstance(issue, dict):
            continue
        normalized = _normalize_issue(issue)
        if normalized is not None:
            found.append(normalized)
    return found


def _normalize_issue(issue: dict[str, Any]) -> dict[str, Any] | None:
    """Normalize one finding, or None when it cannot be located in a file.

    A finding with no line cannot be anchored to anything the patch validator
    can check, so it is dropped here rather than sent to a model that would
    have to guess where it belongs.
    """

    key = str(issue.get("key") or "").strip()
    path = component_path(issue.get("component"))
    line = issue.get("line")
    if not key or not path or not isinstance(line, int) or isinstance(line, bool):
        return None
    return {
        "key": key,
        "path": path,
        "line": line,
        "rule": str(issue.get("rule") or "").strip(),
        "severity": str(issue.get("severity") or "").strip(),
        "type": str(issue.get("type") or "").strip(),
        "effort": str(issue.get("effort") or "").strip(),
        "message": " ".join(str(issue.get("message") or "").split())[:400],
    }


def component_path(component: Any) -> str:
    """Return the repository-relative path from a SonarQube component key.

    Inputs: a component key, which SonarQube formats as `project:path`.
    Outputs: the path, or "" when the key names a project rather than a file.
    """

    text = str(component or "")
    _, separator, path = text.partition(":")
    return path.strip() if separator else ""


INCIDENT_PREFIX = "sonar/"
# The console opens one finding in place, with the rule, the effort estimate
# and the offending lines highlighted.
_ISSUE_PATH = "/project/issues?resolved=false&id={project}&open={key}"


def issue_url(incident_id: str, ui_url: str) -> str:
    """Build the console link for the finding an incident came from.

    Inputs: an incident id shaped `sonar/<project>/<key>`, and the SonarQube
    base url. Outputs: the deep link, or "" for any incident that did not come
    from a sweep - a build-driven repair has no finding to point at.

    Parsed from the incident id rather than threaded through the flow: the id
    already carries both halves, and inventing a second path for them to travel
    is a second thing that can disagree with the first.
    """

    base = str(ui_url or "").rstrip("/")
    text = str(incident_id or "")
    if not base or not text.startswith(INCIDENT_PREFIX):
        return ""
    project, separator, key = text[len(INCIDENT_PREFIX) :].partition("/")
    if not separator or not project.strip() or not key.strip():
        return ""
    return base + _ISSUE_PATH.format(project=project.strip(), key=key.strip())


def _types(cfg: dict) -> tuple[str, ...]:
    """Return the finding types this deployment allows, defaulting to all."""

    raw = cfg.get("sonar_types")
    names = {str(item).strip().upper() for item in (raw or []) if str(item).strip()}
    if not names:
        return ALL_ISSUE_TYPES
    return tuple(item for item in ALL_ISSUE_TYPES if item in names) or ALL_ISSUE_TYPES


def _severities(cfg: dict) -> tuple[str, ...]:
    """Return the severities this deployment restricts to, if any."""

    raw = cfg.get("sonar_severities")
    return tuple(str(item).strip().upper() for item in (raw or []) if str(item).strip())


def _headers(cfg: dict) -> dict[str, str]:
    """Build the basic-auth header SonarQube expects for a user token."""

    token = str(cfg.get("sonar_token") or "")
    encoded = base64.b64encode(f"{token}:".encode("utf-8")).decode("ascii")
    return {"Authorization": f"Basic {encoded}", "Accept": "application/json"}


def _page_size(cfg: dict) -> int:
    """Clamp the page size to what the API accepts."""

    try:
        size = int(cfg.get("sonar_page_size") or _DEFAULT_PAGE_SIZE)
    except (TypeError, ValueError):
        return _DEFAULT_PAGE_SIZE
    return max(1, min(size, _MAX_PAGE_SIZE))


def _timeout(cfg: dict) -> float:
    """Return the request timeout, defaulting when unset or unparseable."""

    try:
        return float(cfg.get("timeout_seconds") or _DEFAULT_TIMEOUT_SECONDS)
    except (TypeError, ValueError):
        return _DEFAULT_TIMEOUT_SECONDS